feat: implement event.srm.operation.completed flow (uc9) (a705b4a6) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/database/repos/qod_sessions.py

+5 −0

Original line number	Diff line number	Diff line
		@@ -18,6 +18,11 @@ class SqlQodSessionRepository(QodSessionRepository):
		row = await self._session.scalar(stmt)
		return QodSessionMapper.to_domain(row) if row is not None else None

		async def get_by_operation_id(self, operation_id: UUID) -> QodSession \| None:
		stmt = select(QodSessionRow).where(QodSessionRow.operation_id == operation_id)
		row = await self._session.scalar(stmt)
		return QodSessionMapper.to_domain(row) if row is not None else None

		async def save(self, qod_session: QodSession) -> QodSession:
		merged = await self._session.merge(QodSessionMapper.to_row(qod_session))
		await self._session.flush()

src/open_exposure_gateway/adapters/databus/nats_adapter.py

+7 −2

Original line number	Diff line number	Diff line
		@@ -9,7 +9,7 @@ from nats.aio.subscription import Subscription
		from pydantic import ValidationError

		from open_exposure_gateway.core.config import NatsSettings
		from open_exposure_gateway.domain.edge_application_management import SRMOperationCompleted
		from open_exposure_gateway.domain.srm_events import SRMOperationCompleted
		from open_exposure_gateway.ports.databus_port import DataBusPort

		logger: structlog.BoundLogger = structlog.get_logger(__name__)
		@@ -78,7 +78,7 @@ class NatsOperationConsumer:
		self,
		client: Client,
		subject: str,
		handler: Callable[[SRMOperationCompleted], Awaitable[None]],
		handler: Callable[[SRMOperationCompleted], Awaitable[None]] \| None = None,
		) -> None:
		self._client = client
		self._subject = subject
		@@ -99,6 +99,11 @@ class NatsOperationConsumer:
		logger.warning("invalid_json", subject=msg.subject)
		return

		if self._handler is None:
		logger.info("operation_completed_received", subject=msg.subject, payload=raw)
		return


		try:
		event = SRMOperationCompleted.model_validate(raw)
		except ValidationError as exc:

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+58 −0

Original line number	Diff line number	Diff line
		@@ -2,6 +2,7 @@ from datetime import datetime, timezone
		from typing import Optional
		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 (
		@@ -21,11 +22,20 @@ from open_exposure_gateway.domain.models import (
		QodSessionState,
		)
		from open_exposure_gateway.domain.quality_on_demand import Subject
		from open_exposure_gateway.domain.srm_events import SRMOperationCompleted
		from open_exposure_gateway.ports.database.operations import OperationRepository
		from open_exposure_gateway.ports.database.qod_sessions import QodSessionRepository
		from open_exposure_gateway.ports.databus_port import DataBusPort
		from open_exposure_gateway.ports.srm_port import SRMClientPort

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

		_OPERATION_COMPLETION_STATUS_MAP: dict[str, OperationStatus] = {
		"completed": OperationStatus.COMPLETED,
		"partially_completed": OperationStatus.PARTIALLY_COMPLETED,
		"failed": OperationStatus.FAILED,
		}


		class QualityOnDemandService:
		def __init__(
		@@ -124,6 +134,54 @@ class QualityOnDemandService:
		qosStatus=QosStatus.REQUESTED,
		)

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

		# Not ours: event.srm.operation.completed is shared across every domain
		# (deploy/terminate and network-capability activate/deactivate all land
		# here). operation_type is how a domain recognizes its own operations.
		if operation.operation_type != OperationType.NETWORK_CAPABILITY:
		return

		status = _OPERATION_COMPLETION_STATUS_MAP[event.status]
		result = None
		if status != OperationStatus.FAILED:
		result = {"instances": [i.model_dump(mode="json") for i in event.instances]}
		await self._operation_repo.save(
		operation.model_copy(
		update={
		"status": status,
		"result": result,
		"error": event.error,
		"completed_at": datetime.fromisoformat(event.completed_at),
		}
		)
		)

		qod_session = await self._qod_session_repo.get_by_operation_id(operation_id)
		if qod_session is None:
		logger.warning(
		"qod_session_completed_for_unknown_operation", operation_id=event.operation_id
		)
		return

		instance = event.instances[0] if event.instances else None
		update: dict[str, object]
		if instance is not None and instance.status == "completed":
		update = {"state": QodSessionState.AVAILABLE, "external_ref": instance.external_ref}
		else:
		update = {"state": QodSessionState.UNAVAILABLE}
		await self._qod_session_repo.save(qod_session.model_copy(update=update))

		async def get_session(
		self,
		session_id: str,

src/open_exposure_gateway/domain/edge_application_management.py

+9 −42

Original line number	Diff line number	Diff line
		from __future__ import annotations

		from enum import StrEnum
		from typing import Any, Literal
		from typing import Any
		from uuid import UUID

		from pydantic import BaseModel, Field, model_validator
		from pydantic import BaseModel, Field

		from open_exposure_gateway.domain.srm_events import (
		SRMCompletedInstance,
		SRMOperationCompleted,
		)

		__all__ = ["SRMCompletedInstance", "SRMOperationCompleted"]


		class Subject(StrEnum):
		@@ -275,46 +282,6 @@ class SRMTerminateCommand(BaseModel):
		terminate: SRMTerminatePayload


		class SRMCompletedInstance(BaseModel):
		service_instance_id: str
		zone_id: str
		status: Literal["completed", "failed"]
		external_ref: str \| None = None
		error: dict[str, Any] \| None = None

		@model_validator(mode="after")
		def _require_error_when_failed(self) -> SRMCompletedInstance:
		if self.status == "failed" and self.error is None:
		raise ValueError("error is required when instance status is failed")
		return self


		class SRMOperationCompleted(BaseModel):
		schema_version: str
		operation_id: str
		status: Literal["completed", "partially_completed", "failed"]
		service_order_id: str \| None = None
		instances: list[SRMCompletedInstance] = []
		metadata: dict[str, Any] \| None = None
		error: dict[str, Any] \| None = None
		correlation_id: str
		completed_at: str

		@model_validator(mode="after")
		def _require_error_when_failed(self) -> SRMOperationCompleted:
		if self.status == "failed" and not self.instances and self.error is None:
		raise ValueError(
		"error is required when status is failed and no instances were produced"
		)
		return self

		@model_validator(mode="after")
		def _require_instances_unless_failed(self) -> SRMOperationCompleted:
		if self.status != "failed" and not self.instances:
		raise ValueError("instances is required when status != failed")
		return self


		class SRMCapabilityEndpoint(BaseModel):
		interface_id: str
		fqdn: str \| None = None

src/open_exposure_gateway/domain/quality_on_demand.py

+1 −0

Original line number	Diff line number	Diff line
		@@ -8,6 +8,7 @@ from pydantic import BaseModel, Field

		class Subject(StrEnum):
		TASK_ACTIVATE = "command.srm.network.capability.activate"
		OPERATION_COMPLETED = "event.srm.operation.completed"


		class NetworkCapabilityTargetDevice(BaseModel):

Admin message