feat: align status metadata with capability_status and route by operation type (c7f0f1be) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/network_capability_condition.py

0 → 100644

+25 −0

Changes for src/open_exposure_gateway/application/services/network_capability_condition.py: 25 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""Shared parsing of async network capability condition metadata on status events.

		SRM emits ``capability_status`` / ``capability_status_info`` (interface-contract §C.4).
		"""

		from __future__ import annotations

		from typing import Any, Literal, TypeGuard

		CapabilityStatus = Literal["AVAILABLE", "UNAVAILABLE"]


		def capability_condition(
		metadata: dict[str, Any] \| None,
		) -> tuple[str \| None, str \| None]:
		md = metadata or {}
		status = md.get("capability_status")
		info = md.get("capability_status_info")
		status_str = status if isinstance(status, str) else None
		info_str = info if isinstance(info, str) else None
		return status_str, info_str


		def is_capability_status(status: str \| None) -> TypeGuard[CapabilityStatus]:
		return status in ("AVAILABLE", "UNAVAILABLE")

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+12 −27

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 12 added lines, 27 removed lines.
Original line number	Diff line number	Diff line
		@@ -22,6 +22,10 @@ from open_exposure_gateway.application.mappers.quality_on_demand_mapper import (
		session_target_from_capability,
		session_target_from_row,
		)
		from open_exposure_gateway.application.services.network_capability_condition import (
		capability_condition,
		is_capability_status,
		)
		from open_exposure_gateway.core.config import (
		DEFAULT_QOD_SERVICE_SPECIFICATION_ID,
		get_settings,
		@@ -423,23 +427,11 @@ class QualityOnDemandService:
		)
		)

		async def handle_completed(self, event: SRMOperationCompleted) -> None:
		async def handle_completed(self, event: SRMOperationCompleted, operation: Operation) -> None:
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")

		operation_id = UUID(event.operation_id)
		operation = await self._operation_repo.get_by_id(operation_id)
		if operation is None:
		logger.warning(
		"operation_completed_for_unknown_operation", operation_id=event.operation_id
		)
		return
		if operation.operation_type not in (
		OperationType.QOD_SESSION,
		OperationType.QOD_SESSION_DEACTIVATE,
		):
		return

		operation_id = operation.operation_id
		status = _OPERATION_COMPLETION_STATUS_MAP[event.status]
		result = None
		if status != OperationStatus.FAILED:
		@@ -530,15 +522,11 @@ class QualityOnDemandService:
		)
		await self._qod_session_repo.save(qod_session.model_copy(update={"state": terminal_state}))

		async def handle_status_changed(self, event: SRMOperationStatus) -> None:
		async def handle_status_changed(self, event: SRMOperationStatus, operation: Operation) -> None:
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")

		operation_id = UUID(event.operation_id)
		operation = await self._operation_repo.get_by_id(operation_id)
		if operation is None or operation.operation_type != OperationType.QOD_SESSION:
		return

		operation_id = operation.operation_id
		qod_session = await self._qod_session_repo.get_by_operation_id(operation_id)
		if (
		qod_session is None
		@@ -557,10 +545,10 @@ class QualityOnDemandService:
		)
		return

		qos_status = (event.metadata or {}).get("qos_status")
		if qos_status not in ("AVAILABLE", "UNAVAILABLE"):
		qos_status, reason = capability_condition(event.metadata)
		if not is_capability_status(qos_status):
		logger.warning(
		"operation_status_unrecognized_qos_status",
		"operation_status_unrecognized_capability_status",
		operation_id=event.operation_id,
		metadata=event.metadata,
		)
		@@ -571,10 +559,7 @@ class QualityOnDemandService:
		)
		status_info: Literal["DURATION_EXPIRED", "NETWORK_TERMINATED"] \| None = None
		if qos_status == "UNAVAILABLE":
		reason = (event.metadata or {}).get("qos_status_info")
		status_info = _QOS_STATUS_INFO_MAP.get(
		reason if isinstance(reason, str) else "", "NETWORK_TERMINATED"
		)
		status_info = _QOS_STATUS_INFO_MAP.get(reason or "", "NETWORK_TERMINATED")

		prior_state = qod_session.state
		update: dict[str, object] = {"state": status}

src/open_exposure_gateway/application/services/traffic_influence_service.py

+12 −30

Changes for src/open_exposure_gateway/application/services/traffic_influence_service.py: 12 added lines, 30 removed lines.
Original line number	Diff line number	Diff line
		@@ -25,6 +25,9 @@ from open_exposure_gateway.application.mappers.traffic_influence_mapper import (
		build_traffic_influence_changed_event,
		build_traffic_influence_response,
		)
		from open_exposure_gateway.application.services.network_capability_condition import (
		capability_condition,
		)
		from open_exposure_gateway.core.config import DEFAULT_TI_SERVICE_SPECIFICATION_ID, get_settings
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		@@ -596,25 +599,11 @@ class TrafficInfluenceService:
		if registration is not None:
		await self._callback_credential_repo.revoke_by_registration_id(registration.id)

		async def handle_completed(self, event: SRMOperationCompleted) -> None:
		async def handle_completed(self, event: SRMOperationCompleted, operation: Operation) -> None:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		operation_id = UUID(event.operation_id)
		operation = await self._operation_repo.get_by_id(operation_id)
		if operation is None:
		logger.warning(
		"operation_completed_for_unknown_operation", operation_id=event.operation_id
		)
		return

		# Not ours: event.srm.operation.completed is shared across every domain.
		if operation.operation_type not in (
		OperationType.TRAFFIC_INFLUENCE,
		OperationType.TRAFFIC_INFLUENCE_DEACTIVATE,
		):
		return

		operation_id = operation.operation_id
		status = _OPERATION_COMPLETION_STATUS_MAP[event.status]
		result = None
		if status != OperationStatus.FAILED:
		@@ -696,27 +685,20 @@ class TrafficInfluenceService:
		if saved.state == RecordTrafficInfluenceState.DELETED:
		await self._forget_callback_credential(saved.operation_id)

		async def handle_status_changed(self, event: SRMOperationStatus) -> None:
		async def handle_status_changed(self, event: SRMOperationStatus, operation: Operation) -> None:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		operation_id = UUID(event.operation_id)
		operation = await self._operation_repo.get_by_id(operation_id)
		if operation is None or operation.operation_type != OperationType.TRAFFIC_INFLUENCE:
		return

		operation_id = operation.operation_id
		traffic_influence = await self._traffic_influence_repo.get_by_operation_id(operation_id)
		if traffic_influence is None:
		return

		# Same backend-driven "capability condition changed" signal QoD's
		# handle_status_changed reads (srm/interface-contract.md §C.1/§C.4 documents
		# only the qos_status example, but the metadata key is generic across network
		# capabilities, not QoD-specific).
		qos_status = (event.metadata or {}).get("qos_status")
		if qos_status not in ("AVAILABLE", "UNAVAILABLE"):
		# Backend-driven capability condition (interface-contract §C.4).
		capability_status, _info = capability_condition(event.metadata)
		if capability_status not in ("AVAILABLE", "UNAVAILABLE"):
		logger.warning(
		"operation_status_unrecognized_qos_status",
		"operation_status_unrecognized_capability_status",
		operation_id=event.operation_id,
		metadata=event.metadata,
		)
		@@ -732,7 +714,7 @@ class TrafficInfluenceService:

		state = (
		RecordTrafficInfluenceState.ACTIVE
		if qos_status == "AVAILABLE"
		if capability_status == "AVAILABLE"
		else RecordTrafficInfluenceState.ERROR
		)
		prior_state = traffic_influence.state

src/open_exposure_gateway/domain/quality_on_demand.py

+1 −1

Changes for src/open_exposure_gateway/domain/quality_on_demand.py: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -140,7 +140,7 @@ class SRMOperationStatus(BaseModel):
		Covers both pre-execution states (`accepted`, `failed_before_start`) and, per §C.3's
		"backend-driven change" case, post-completion capability condition changes (e.g. a QoD
		session dropped by the network) reusing the original realization's `operation_id`.
		Backend-specific status (e.g. `qos_status`) lives in `metadata`, never in `state`.
		Backend-specific status (e.g. `capability_status`) lives in `metadata`, never in `state`.
		"""

		schema_version: str

src/open_exposure_gateway/main.py

+68 −31

Changes for src/open_exposure_gateway/main.py: 68 added lines, 31 removed lines.
Original line number	Diff line number	Diff line
		@@ -101,6 +101,7 @@ from open_exposure_gateway.domain.edge_application_management import (
		SRMOperationCompleted,
		Subject,
		)
		from open_exposure_gateway.domain.models.operations.enums import OperationType
		from open_exposure_gateway.domain.quality_on_demand import SRMOperationStatus
		from open_exposure_gateway.domain.quality_on_demand import Subject as QodSubject
		from open_exposure_gateway.ports.callback_delivery_port import CallbackDeliveryPort
		@@ -111,6 +112,8 @@ from open_exposure_gateway.ports.traffic_influence_callback_port import (
		TrafficInfluenceCallbackDeliveryPort,
		)

		logger: structlog.BoundLogger = structlog.get_logger(__name__)


		def _build_eam_service(
		session: AsyncSession,
		@@ -136,16 +139,22 @@ _OPERATION_LOOKUP_RETRY_DELAYS: tuple[float, ...] = (0.05, 0.1, 0.2, 0.4)

		async def _wait_for_operation(
		session_maker: async_sessionmaker[AsyncSession], operation_id: str
		) -> None:
		) -> UUID \| None:
		"""Parse ``operation_id`` and briefly retry until the row is visible (publish-before-commit).

		Returns the parsed UUID, or ``None`` when the id is not a UUID (already logged).
		"""
		try:
		parsed_id = UUID(operation_id)
		except ValueError:
		return
		logger.warning("operation_event_invalid_operation_id", operation_id=operation_id)
		return None
		for delay in _OPERATION_LOOKUP_RETRY_DELAYS:
		async with session_maker() as session:
		if await SqlOperationRepository(session).get_by_id(parsed_id) is not None:
		return
		return parsed_id
		await asyncio.sleep(delay)
		return parsed_id


		def _build_operation_completed_handler(
		@@ -154,7 +163,8 @@ def _build_operation_completed_handler(
		callback_delivery_port: CallbackDeliveryPort,
		) -> Callable[[SRMOperationCompleted], Awaitable[None]]:
		async def handle(event: SRMOperationCompleted) -> None:
		await _wait_for_operation(session_maker, event.operation_id)
		if await _wait_for_operation(session_maker, event.operation_id) is None:
		return
		async with session_maker() as session:
		try:
		service = _build_eam_service(session, srm_client, callback_delivery_port)
		@@ -173,7 +183,8 @@ def _build_eam_operation_status_handler(
		callback_delivery_port: CallbackDeliveryPort,
		) -> Callable[[SRMOperationStatus], Awaitable[None]]:
		async def handle(event: SRMOperationStatus) -> None:
		await _wait_for_operation(session_maker, event.operation_id)
		if await _wait_for_operation(session_maker, event.operation_id) is None:
		return
		async with session_maker() as session:
		try:
		service = _build_eam_service(session, srm_client, callback_delivery_port)
		@@ -360,31 +371,44 @@ def _build_traffic_influence_service(
		)


		def _build_qod_operation_completed_handler(
		def _build_shared_operation_completed_handler(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		qod_callback_client: QodCallbackDeliveryPort,
		traffic_influence_callback_client: TrafficInfluenceCallbackDeliveryPort,
		) -> Callable[[SRMOperationCompleted], Awaitable[None]]:
		"""Dispatches one event.srm.operation.completed to every non-EAM domain that might
		own it (QoD, Traffic Influence).
		"""Routes ``event.srm.operation.completed`` to QoD or TI by ``operation_type``.

		Each service's handle_completed self-filters by operation_type and no-ops if the
		event isn't its own, so it's safe to call all of them from a single consumer --
		this is a second subscription for a subject shared with EAM's own consumer (srm/
		interface-contract.md §B.5/§D.3), not one consumer per domain.
		Second consumer on a subject also used by EAM (srm/interface-contract.md §B.5/§D.3);
		deploy/scale/terminate (and any other type) are ignored here.
		"""

		async def handle(event: SRMOperationCompleted) -> None:
		await _wait_for_operation(session_maker, event.operation_id)
		operation_id = await _wait_for_operation(session_maker, event.operation_id)
		if operation_id is None:
		return
		async with session_maker() as session:
		try:
		qod_service = _build_qod_service(session, srm_client, qod_callback_client)
		traffic_influence_service = _build_traffic_influence_service(
		session, srm_client, traffic_influence_callback_client
		operation = await SqlOperationRepository(session).get_by_id(operation_id)
		if operation is None:
		logger.warning(
		"operation_event_for_unknown_operation",
		operation_id=event.operation_id,
		)
		await qod_service.handle_completed(event)
		await traffic_influence_service.handle_completed(event)
		return
		match operation.operation_type:
		case OperationType.QOD_SESSION \| OperationType.QOD_SESSION_DEACTIVATE:
		await _build_qod_service(
		session, srm_client, qod_callback_client
		).handle_completed(event, operation)
		case (
		OperationType.TRAFFIC_INFLUENCE \| OperationType.TRAFFIC_INFLUENCE_DEACTIVATE
		):
		await _build_traffic_influence_service(
		session, srm_client, traffic_influence_callback_client
		).handle_completed(event, operation)
		case _:
		return
		await session.commit()
		except Exception:
		await session.rollback()
		@@ -393,29 +417,42 @@ def _build_qod_operation_completed_handler(
		return handle


		def _build_qod_operation_status_handler(
		def _build_shared_operation_status_handler(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		qod_callback_client: QodCallbackDeliveryPort,
		traffic_influence_callback_client: TrafficInfluenceCallbackDeliveryPort,
		) -> Callable[[SRMOperationStatus], Awaitable[None]]:
		"""Dispatches one event.srm.operation.status to every domain that might own it.
		"""Routes ``event.srm.operation.status`` to QoD or TI by ``operation_type``.

		Same shared-subject dispatch pattern as _build_qod_operation_completed_handler: each
		service's handle_status_changed self-filters by operation_type and no-ops if the
		event isn't its own.
		Only activate operations: condition changes reuse the activate ``operation_id``
		(interface-contract §C.3/§C.4). Deactivate and EAM types are ignored here.
		"""

		async def handle(event: SRMOperationStatus) -> None:
		await _wait_for_operation(session_maker, event.operation_id)
		operation_id = await _wait_for_operation(session_maker, event.operation_id)
		if operation_id is None:
		return
		async with session_maker() as session:
		try:
		qod_service = _build_qod_service(session, srm_client, qod_callback_client)
		traffic_influence_service = _build_traffic_influence_service(
		session, srm_client, traffic_influence_callback_client
		operation = await SqlOperationRepository(session).get_by_id(operation_id)
		if operation is None:
		logger.warning(
		"operation_event_for_unknown_operation",
		operation_id=event.operation_id,
		)
		await qod_service.handle_status_changed(event)
		await traffic_influence_service.handle_status_changed(event)
		return
		match operation.operation_type:
		case OperationType.QOD_SESSION:
		await _build_qod_service(
		session, srm_client, qod_callback_client
		).handle_status_changed(event, operation)
		case OperationType.TRAFFIC_INFLUENCE:
		await _build_traffic_influence_service(
		session, srm_client, traffic_influence_callback_client
		).handle_status_changed(event, operation)
		case _:
		return
		await session.commit()
		except Exception:
		await session.rollback()
		@@ -604,7 +641,7 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		operation_completed_consumer = NatsOperationConsumer(
		client=publisher.client,
		subject=Subject.OPERATION_COMPLETED,
		handler=_build_qod_operation_completed_handler(
		handler=_build_shared_operation_completed_handler(
		session_maker, srm_client, qod_callback_client, traffic_influence_callback_client
		),
		)
		@@ -614,7 +651,7 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		operation_status_consumer = NatsOperationStatusConsumer(
		client=publisher.client,
		subject=QodSubject.OPERATION_STATUS,
		handler=_build_qod_operation_status_handler(
		handler=_build_shared_operation_status_handler(
		session_maker, srm_client, qod_callback_client, traffic_influence_callback_client
		),
		)