feat: handle event.srm.operation.status for backend-driven stage changes (0f266031) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/traffic_influence_service.py

+39 −1

Original line number	Diff line number	Diff line
		@@ -36,7 +36,7 @@ from open_exposure_gateway.domain.models import (
		from open_exposure_gateway.domain.models import (
		TrafficInfluenceState as RecordTrafficInfluenceState,
		)
		from open_exposure_gateway.domain.srm_events import SRMOperationCompleted
		from open_exposure_gateway.domain.srm_events import SRMOperationCompleted, SRMOperationStatus
		from open_exposure_gateway.domain.traffic_influence import Subject
		from open_exposure_gateway.ports.database.callbacks import CallbackRegistrationRepository
		from open_exposure_gateway.ports.database.operations import OperationRepository
		@@ -313,6 +313,44 @@ class TrafficInfluenceService:
		traffic_influence.model_copy(update=deactivate_update)
		)

		async def handle_status_changed(self, event: SRMOperationStatus) -> 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

		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"):
		logger.warning(
		"operation_status_unrecognized_qos_status",
		operation_id=event.operation_id,
		metadata=event.metadata,
		)
		return

		state = (
		RecordTrafficInfluenceState.ACTIVE
		if qos_status == "AVAILABLE"
		else RecordTrafficInfluenceState.ERROR
		)
		await self._traffic_influence_repo.save(
		traffic_influence.model_copy(update={"state": state})
		)

		# Delivery of onTrafficInfluenceChanged is a separate, not-yet-built concern
		# (same scoping as the subscriptionRequest note in create_traffic_influence).

		async def get_traffic_influence(
		self,
		traffic_influence_id: str,

src/open_exposure_gateway/domain/traffic_influence.py

+1 −0

Original line number	Diff line number	Diff line
		@@ -12,6 +12,7 @@ class Subject(StrEnum):
		TASK_ACTIVATE = "command.srm.network.capability.activate"
		TASK_DEACTIVATE = "command.srm.network.capability.deactivate"
		OPERATION_COMPLETED = "event.srm.operation.completed"
		OPERATION_STATUS = "event.srm.operation.status"


		class NetworkCapabilityTargetApplicationReference(BaseModel):

src/open_exposure_gateway/main.py

+23 −7

Original line number	Diff line number	Diff line
		@@ -111,16 +111,17 @@ def _build_qod_service(
		)


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

		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 the one subscription for a subject shared across domains (srm/
		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.
		"""

		@@ -148,11 +149,24 @@ def _build_qod_operation_status_handler(
		srm_client: SRMClientPort,
		qod_callback_client: QodCallbackDeliveryPort,
		) -> Callable[[SRMOperationStatus], Awaitable[None]]:
		"""Dispatches one event.srm.operation.status to every domain that might own it.

		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.
		"""

		async def handle(event: SRMOperationStatus) -> None:
		async with session_maker() as session:
		try:
		service = _build_qod_service(session, srm_client, qod_callback_client)
		await service.handle_status_changed(event)
		qod_service = _build_qod_service(session, srm_client, qod_callback_client)
		traffic_influence_service = TrafficInfluenceService(
		srm_client=srm_client,
		operation_repo=SqlOperationRepository(session),
		traffic_influence_repo=SqlTrafficInfluenceRepository(session),
		)
		await qod_service.handle_status_changed(event)
		await traffic_influence_service.handle_status_changed(event)
		await session.commit()
		except Exception:
		await session.rollback()
		@@ -235,14 +249,16 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		eam_operation_consumer = NatsOperationConsumer(
		client=publisher.client,
		subject=Subject.OPERATION_COMPLETED,
		handler=_build_operation_completed_handler(session_maker, srm_client, HttpCallbackClient()),
		handler=_build_operation_completed_handler(
		session_maker, srm_client, HttpCallbackClient()
		),
		)
		await eam_operation_consumer.start()

		operation_completed_consumer = NatsOperationConsumer(
		client=publisher.client,
		subject=Subject.OPERATION_COMPLETED,
		handler=_build_operation_completed_handler(
		handler=_build_qod_operation_completed_handler(
		session_maker, srm_client, qod_callback_client
		),
		)

Admin message