Commit 679d6fda authored by Dimitrios Gogos's avatar Dimitrios Gogos
Browse files

refactor: update QoD service specification handling and integrate network capability retrieval

parent a1f88b03

.env.example

+1 −0
Original line number Diff line number Diff line
@@ -23,3 +23,4 @@ 
# OBSERVABILITY_SETTINGS__LOG_LEVEL="INFO"



# CALLBACK_SETTINGS__TIMEOUT=10.0

# QOD_SETTINGS__SERVICE_SPECIFICATION_ID="7608e902-b927-559f-b448-e7e9061dfa5c"

src/open_exposure_gateway/adapters/http/srm_client.py

+8 −8
Original line number Diff line number Diff line
@@ -4,9 +4,6 @@ from uuid import UUID 
import httpx

import structlog



from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (

    SessionInfo,

)

from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.core.exceptions import (

    DownstreamServiceException,
@@ -18,6 +15,7 @@ from open_exposure_gateway.domain.edge_application_management import ( 
    SRMServiceInstance,

    SRMZone,

)

from open_exposure_gateway.domain.quality_on_demand import SRMNetworkCapability



logger = structlog.get_logger(__name__)
@@ -124,14 +122,16 @@ class SRMClient: 
        )

        return [SRMZone.model_validate(z) for z in data]



    async def get_qod_session(

    async def get_network_capability(

        self,

        session_id: str,

        service_instance_id: str,

        x_correlator: str | None = None,

    ) -> SessionInfo:

    ) -> SRMNetworkCapability:

        headers = {"x-correlator": x_correlator} if x_correlator else None

        data = await self._request("GET", f"/sessions/{session_id}", headers=headers)

        return SessionInfo.model_validate(data)

        data = await self._request(

            "GET", f"/internal/network-capabilities/{service_instance_id}", headers=headers

        )

        return SRMNetworkCapability.model_validate(data)



    async def get_apps(self, x_correlator: str | None = None) -> list[SRMCatalogPayload]:

        headers = {"x-correlator": x_correlator} if x_correlator else None

src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py

+59 −1
Original line number Diff line number Diff line 
from datetime import datetime, timezone

from typing import Literal

from uuid import UUID, uuid4



from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import CreateSession

from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (

    ApplicationServer,

    CreateSession,

    Device,

    DeviceIpv4Addr,

    QosStatus,

    SessionInfo,

)

from open_exposure_gateway.domain.models import QodSession

from open_exposure_gateway.domain.quality_on_demand import (

    EventQosStatusChangedData,

    NetworkCapabilityDeactivateTarget,
@@ -11,6 +20,7 @@ from open_exposure_gateway.domain.quality_on_demand import ( 
    NetworkCapabilityTargetApplicationServer,

    NetworkCapabilityTargetDevice,

    QosStatusChangedCloudEvent,

    SRMNetworkCapability,

    SRMNetworkCapabilityActivateCommand,

    SRMNetworkCapabilityDeactivateCommand,

)
@@ -22,6 +32,7 @@ def build_activate_command( 
    correlation_id: str,

    requested_at: str,

    service_specification_id: UUID,

    service_instance_id: UUID,

    app_provider_id: str,

) -> SRMNetworkCapabilityActivateCommand:

    device = request.device
@@ -45,6 +56,7 @@ def build_activate_command( 
        correlation_id=correlation_id,

        requested_at=requested_at,

        app_provider_id=app_provider_id,

        service_instance_id=str(service_instance_id),

        service_specification_id=str(service_specification_id),

        network_capability=NetworkCapabilityPayload(

            target=target,
@@ -54,6 +66,52 @@ def build_activate_command( 
    )





def build_session_info(qod_session: QodSession, capability: SRMNetworkCapability) -> SessionInfo:

    """Combines `qod_sessions` (identity, ownership, cached `qosStatus`) with SRM's live

    capability detail (target device/application server) into the CAMARA response -- the

    read-side counterpart of `build_activate_command`, and the boundary that keeps SRM's

    internal shape from leaking onto the CAMARA surface directly.

    """

    device_target = capability.parameters_snapshot.target.device

    server_target = capability.parameters_snapshot.target.application_server



    ipv4_address = None

    if device_target.ipv4:

        # Only the bare address survives on SRM's side (build_activate_command never sends

        # privateAddress/publicPort), so this can't fully round-trip a CAMARA DeviceIpv4Addr.

        # model_construct bypasses DeviceIpv4Addr's "publicAddress + one of private/port"

        # validator, which exists to constrain client input, not this internal reconstruction.

        # Device.model_construct is required alongside it: pydantic revalidates nested model

        # fields against their own validators even when a pre-built instance is passed in, so

        # the bypass has to hold at every level enclosing the incomplete address.

        ipv4_address = DeviceIpv4Addr.model_construct(

            publicAddress=device_target.ipv4, privateAddress=None, publicPort=None

        )



    device = Device.model_construct(

        phoneNumber=device_target.phone_number,

        networkAccessIdentifier=device_target.network_access_id,

        ipv4Address=ipv4_address,

        ipv6Address=device_target.ipv6,

    )

    application_server = ApplicationServer(

        ipv4Address=server_target.ipv4,

        ipv6Address=server_target.ipv6,

    )



    started_at = int((qod_session.created_at or datetime.now(timezone.utc)).timestamp())

    return SessionInfo(

        sessionId=qod_session.session_id,

        device=device,

        applicationServer=application_server,

        qosProfile=qod_session.qos_profile,

        duration=qod_session.duration_seconds,

        startedAt=started_at,

        expiresAt=started_at + qod_session.duration_seconds,

        qosStatus=QosStatus(qod_session.state.value),

    )





def build_deactivate_command(

    external_ref: str,

    operation_id: UUID,

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+22 −4
Original line number Diff line number Diff line
@@ -14,8 +14,12 @@ from open_exposure_gateway.application.mappers.quality_on_demand_mapper import ( 
    build_activate_command,

    build_deactivate_command,

    build_qos_status_changed_event,

    build_session_info,

)

from open_exposure_gateway.core.config import (

    DEFAULT_QOD_SERVICE_SPECIFICATION_ID,

    get_settings,

)

from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.core.exceptions import DownstreamServiceException, NotFoundException

from open_exposure_gateway.domain.edge_application_management import SRMOperationCompleted

from open_exposure_gateway.domain.models import (
@@ -57,6 +61,7 @@ class QualityOnDemandService: 
        callback_registration_repo: Optional[CallbackRegistrationRepository] = None,

        callback_delivery_repo: Optional[CallbackDeliveryRepository] = None,

        callback_delivery_port: Optional[QodCallbackDeliveryPort] = None,

        service_specification_id: UUID = DEFAULT_QOD_SERVICE_SPECIFICATION_ID,

    ) -> None:

        self.srm_client = srm_client

        self._publisher = publisher
@@ -65,6 +70,7 @@ class QualityOnDemandService: 
        self._callback_registration_repo = callback_registration_repo

        self._callback_delivery_repo = callback_delivery_repo

        self._callback_delivery_port = callback_delivery_port

        self._service_specification_id = service_specification_id



    def _new_operation_metadata(self, x_correlator: Optional[str]) -> tuple[UUID, str, str]:

        operation_id = uuid4()
@@ -92,7 +98,10 @@ class QualityOnDemandService: 


        operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)

        session_id = uuid4()

        service_specification_id = uuid4()

        # The platform-seeded QoD specification, identical on every session (ADR-0035).

        # Stored per row so the offering a session was provisioned under stays auditable

        # once per-offering specifications exist.

        service_specification_id = self._service_specification_id



        command = build_activate_command(

            request=request,
@@ -100,6 +109,7 @@ class QualityOnDemandService: 
            correlation_id=correlation_id,

            requested_at=requested_at,

            service_specification_id=service_specification_id,

            service_instance_id=session_id,

            app_provider_id=app_provider_id,

        )
@@ -318,10 +328,18 @@ class QualityOnDemandService: 
        session_id: str,

        x_correlator: Optional[str] = None,

    ) -> SessionInfo:

        return await self.srm_client.get_qod_session(

            session_id=session_id,

        if self._qod_session_repo is None:

            raise RuntimeError("QodSession repository is not available")



        qod_session = await self._qod_session_repo.get_by_id(UUID(session_id))

        if qod_session is None:

            raise NotFoundException(message=f"Session {session_id} not found")



        capability = await self.srm_client.get_network_capability(

            service_instance_id=session_id,

            x_correlator=x_correlator,

        )

        return build_session_info(qod_session, capability)



    async def delete_session(

        self,

src/open_exposure_gateway/core/config.py

+18 −0
Original line number Diff line number Diff line 
from functools import lru_cache

from uuid import UUID



from pydantic import BaseModel, HttpUrl

from pydantic_settings import BaseSettings, SettingsConfigDict



# The platform seeds one QoD service specification at deployment (ADR-0035); every

# activate carries its id. There is no CAMARA registration step to resolve it from,

# so the id cannot be derived per request -- minting one would reference no catalog

# row and SRM would answer failed_before_start. This is the well-known id used when

# nobody sets one at deploy time; SRM's topology bootstrap must seed the same value.



DEFAULT_QOD_SERVICE_SPECIFICATION_ID = UUID("7608e902-b927-559f-b448-e7e9061dfa5c")





class SRMSettings(BaseModel):

    base_url: HttpUrl = HttpUrl("http://localhost:8081")
@@ -29,6 +38,14 @@ class CallbackSettings(BaseModel): 
    timeout: float = 10.0





class QodSettings(BaseModel):

    service_specification_id: UUID = DEFAULT_QOD_SERVICE_SPECIFICATION_ID



    @property

    def uses_default_service_specification_id(self) -> bool:

        return self.service_specification_id == DEFAULT_QOD_SERVICE_SPECIFICATION_ID





class Settings(BaseSettings):

    model_config = SettingsConfigDict(

        env_file=".env",
@@ -49,6 +66,7 @@ class Settings(BaseSettings): 
    nats_settings: NatsSettings = NatsSettings()

    observability_settings: ObservabilitySettings = ObservabilitySettings()

    callback_settings: CallbackSettings = CallbackSettings()

    qod_settings: QodSettings = QodSettings()





@lru_cache