feat(qod): move qod service and mapper to v1.1.0 (bbce007a) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py

+26 −7

Changes for src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py: 26 added lines, 7 removed lines.
Original line number	Diff line number	Diff line
		from datetime import datetime, timezone
		from datetime import datetime, timedelta, timezone
		from typing import Literal
		from uuid import UUID, uuid4

		from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (
		from open_exposure_gateway.api.camara.quality_on_demand.v1_1_0.schemas import (
		SOURCE_API,
		SOURCE_API_VERSION,
		ApplicationServer,
		CreateSession,
		Device,
		DeviceIpv4Addr,
		DeviceResponse,
		PortsSpec,
		QosStatus,
		SessionInfo,
		@@ -46,8 +47,20 @@ def _to_capability_ports(spec: PortsSpec \| None) -> NetworkCapabilityPorts \| Non
		)


		def build_device_response(device: Device) -> DeviceResponse \| None:
		"""CAMARA returns exactly one identifier: phone number, then IPv4, then IPv6."""
		if device.phoneNumber is not None:
		return DeviceResponse.model_construct(phoneNumber=device.phoneNumber)
		if device.ipv4Address is not None:
		return DeviceResponse.model_construct(ipv4Address=device.ipv4Address)
		if device.ipv6Address is not None:
		return DeviceResponse.model_construct(ipv6Address=device.ipv6Address)
		return None


		def build_activate_command(
		request: CreateSession,
		device: Device,
		operation_id: UUID,
		correlation_id: str,
		requested_at: str,
		@@ -55,7 +68,6 @@ def build_activate_command(
		service_instance_id: UUID,
		app_provider_id: str,
		) -> SRMNetworkCapabilityActivateCommand:
		device = request.device
		application_server = request.applicationServer

		target = NetworkCapabilityTarget(
		@@ -108,12 +120,14 @@ def build_session_info(qod_session: QodSession, capability: SRMNetworkCapability
		publicPort=device_target.ipv4_port,
		)

		device = Device.model_construct(
		device = build_device_response(
		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,
		@@ -129,7 +143,12 @@ def build_session_info(qod_session: QodSession, capability: SRMNetworkCapability
		else None
		)

		started_at = int((qod_session.created_at or datetime.now(timezone.utc)).timestamp())
		qos_status = QosStatus(qod_session.state.value)
		started_at: datetime \| None = None
		expires_at: datetime \| None = None
		if qos_status != QosStatus.REQUESTED:
		started_at = qod_session.created_at or datetime.now(timezone.utc)
		expires_at = started_at + timedelta(seconds=qod_session.duration_seconds)
		return SessionInfo(
		sessionId=qod_session.session_id,
		device=device,
		@@ -139,8 +158,8 @@ def build_session_info(qod_session: QodSession, capability: SRMNetworkCapability
		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),
		expiresAt=expires_at,
		qosStatus=qos_status,
		)

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+36 −15

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 36 added lines, 15 removed lines.
Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@ from uuid import UUID, uuid4
		import structlog
		from pydantic import BaseModel

		from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (
		from open_exposure_gateway.api.camara.quality_on_demand.v1_1_0.schemas import (
		CreateSession,
		QosStatus,
		SessionInfo,
		@@ -13,6 +13,7 @@ from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (
		from open_exposure_gateway.application.mappers.quality_on_demand_mapper import (
		build_activate_command,
		build_deactivate_command,
		build_device_response,
		build_qos_status_changed_event,
		build_session_info,
		)
		@@ -23,7 +24,10 @@ from open_exposure_gateway.core.config import (
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		DownstreamServiceException,
		ErrorCode,
		NotFoundException,
		NotImplementedException,
		UnprocessableEntityException,
		)
		from open_exposure_gateway.domain.edge_application_management import SRMOperationCompleted
		from open_exposure_gateway.domain.models import (
		@@ -63,6 +67,8 @@ _TERMINAL_STATES = frozenset(
		{QodSessionState.DELETION_REQUESTED, QodSessionState.DELETED, QodSessionState.ERROR}
		)

		_QOS_STATUS_CHANGED_EVENT_TYPE = "org.camaraproject.quality-on-demand.v1.qos-status-changed"


		class QualityOnDemandService:
		def __init__(
		@@ -115,6 +121,26 @@ class QualityOnDemandService:
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")

		device = request.device
		if device is None:
		raise UnprocessableEntityException(
		error_code=ErrorCode.MISSING_IDENTIFIER,
		message="The device cannot be identified.",
		)
		if device.networkAccessIdentifier is not None:
		raise UnprocessableEntityException(
		error_code=ErrorCode.UNSUPPORTED_IDENTIFIER,
		message="The identifier provided is not supported.",
		)

		# TODO: remove once callback credentials can be stored.
		if request.sinkCredential is not None:
		raise NotImplementedException(
		"sinkCredential is not supported in this release: the gateway cannot "
		"yet store callback credentials, so an authenticated sink would "
		"receive no notifications"
		)

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		session_id = uuid4()
		# The platform-seeded QoD specification, identical on every session (ADR-0035).
		@@ -124,6 +150,7 @@ class QualityOnDemandService:

		command = build_activate_command(
		request=request,
		device=device,
		operation_id=operation_id,
		correlation_id=correlation_id,
		requested_at=requested_at,
		@@ -168,7 +195,7 @@ class QualityOnDemandService:
		)
		)

		if request.webhook is not None:
		if request.sink is not None:
		if self._callback_registration_repo is None:
		raise RuntimeError("CallbackRegistrationRepository is not available")
		await self._callback_registration_repo.save(
		@@ -177,13 +204,9 @@ class QualityOnDemandService:
		operation_id=operation_id,
		tenant_id=tenant_id,
		api_family="quality-on-demand",
		sink=request.webhook.notificationUrl,
		event_types=["org.camaraproject.qod.v0.qos-status-changed"],
		sink_credential_ref=(
		f"secret://oeg/{operation_id}/notification-auth-token"
		if request.webhook.notificationAuthToken
		else None
		),
		sink=request.sink,
		event_types=[_QOS_STATUS_CHANGED_EVENT_TYPE],
		sink_credential_ref=None,
		expires_at=None,
		)
		)
		@@ -194,18 +217,16 @@ class QualityOnDemandService:
		"Failed to publish QoD session activation",
		)

		started_at = int(datetime.now(timezone.utc).timestamp())
		# startedAt/expiresAt stay unset until the session leaves REQUESTED.
		return SessionInfo(
		sessionId=session_id,
		device=request.device,
		device=build_device_response(device),
		applicationServer=request.applicationServer,
		devicePorts=request.devicePorts,
		applicationServerPorts=request.applicationServerPorts,
		qosProfile=request.qosProfile,
		webhook=request.webhook,
		sink=request.sink,
		duration=request.duration,
		startedAt=started_at,
		expiresAt=started_at + request.duration,
		qosStatus=QosStatus.REQUESTED,
		)

		@@ -225,7 +246,7 @@ class QualityOnDemandService:
		if self._callback_delivery_port is None or self._callback_delivery_repo is None:
		raise RuntimeError("CallbackDeliveryPort/CallbackDeliveryRepository is not available")

		source = f"{get_settings().public_base_url}/qod/v0/sessions/{session_id}"
		source = f"{get_settings().public_base_url}/quality-on-demand/v1/sessions/{session_id}"
		cloud_event = build_qos_status_changed_event(
		session_id=session_id,
		qos_status=qos_status,