Merge branch 'refactor/qod' into 'develop' (f803ad2b) · Commits · OOP / code / open-exposure-gateway

.env.example

+4 −0

Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@
		# DEBUG=false
		# HOST="0.0.0.0"
		# PORT=8080
		# PUBLIC_BASE_URL="http://localhost:8080"

		# SRM_SETTINGS__BASE_URL="http://localhost:8081"
		# SRM_SETTINGS__TIMEOUT=10.0
		@@ -20,3 +21,6 @@
		# NATS_SETTINGS__MAX_RECONNECT_ATTEMPTS=3

		# 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/database/mappers.py

+34 −0

Original line number	Diff line number	Diff line
		@@ -4,6 +4,7 @@ from open_exposure_gateway.adapters.database.sql import (
		CallbackDeliveryRow,
		CallbackRegistrationRow,
		OperationRow,
		QodSessionRow,
		)
		from open_exposure_gateway.domain.models import (
		AppInstance,
		@@ -11,6 +12,7 @@ from open_exposure_gateway.domain.models import (
		CallbackDelivery,
		CallbackRegistration,
		Operation,
		QodSession,
		)


		@@ -108,6 +110,38 @@ class AppInstanceMapper:
		)


		class QodSessionMapper:
		@staticmethod
		def to_domain(row: QodSessionRow) -> QodSession:
		return QodSession(
		session_id=row.session_id,
		operation_id=row.operation_id,
		service_specification_id=row.service_specification_id,
		qos_profile=row.qos_profile,
		duration_seconds=row.duration_seconds,
		state=row.state,
		external_ref=row.external_ref,
		device_ports=row.device_ports,
		application_server_ports=row.application_server_ports,
		created_at=row.created_at,
		updated_at=row.updated_at,
		)

		@staticmethod
		def to_row(domain: QodSession) -> QodSessionRow:
		return QodSessionRow(
		session_id=domain.session_id,
		operation_id=domain.operation_id,
		service_specification_id=domain.service_specification_id,
		qos_profile=domain.qos_profile,
		duration_seconds=domain.duration_seconds,
		state=domain.state,
		external_ref=domain.external_ref,
		device_ports=domain.device_ports,
		application_server_ports=domain.application_server_ports,
		)


		class CallbackRegistrationMapper:
		@staticmethod
		def to_domain(row: CallbackRegistrationRow) -> CallbackRegistration:

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

0 → 100644

+32 −0

Original line number	Diff line number	Diff line
		from uuid import UUID

		from sqlalchemy import select
		from sqlalchemy.ext.asyncio import AsyncSession

		from open_exposure_gateway.adapters.database.mappers import QodSessionMapper
		from open_exposure_gateway.adapters.database.sql import QodSessionRow
		from open_exposure_gateway.domain.models import QodSession
		from open_exposure_gateway.ports.database.qod_sessions import QodSessionRepository


		class SqlQodSessionRepository(QodSessionRepository):
		def __init__(self, session: AsyncSession) -> None:
		self._session = session

		async def get_by_id(self, session_id: UUID) -> QodSession \| None:
		stmt = select(QodSessionRow).where(QodSessionRow.session_id == session_id)
		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()
		saved = await self.get_by_id(merged.session_id)
		if saved is None:
		raise RuntimeError("Saved QoD session could not be reloaded")
		return saved

src/open_exposure_gateway/adapters/database/sql.py

+18 −0

Original line number	Diff line number	Diff line
		@@ -26,6 +26,7 @@ from open_exposure_gateway.domain.models.operations.enums import (
		OperationStatus,
		OperationType,
		)
		from open_exposure_gateway.domain.models.qod_sessions.enums import QodSessionState
		from open_exposure_gateway.domain.models.registration.enums import (
		AppRegistrationStatus,
		PackageType,
		@@ -184,3 +185,20 @@ class AppInstanceRow(AuditedMixin, Base):
		)
		edge_cloud_zone_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), nullable=False)
		state: Mapped[AppInstanceState] = mapped_column(_enum_type(AppInstanceState), nullable=False)


		class QodSessionRow(AuditedMixin, Base):
		__tablename__ = "qod_sessions"
		__table_args__ = (Index("idx_qod_sessions_operation", "operation_id"),)

		session_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)
		operation_id: Mapped[UUID] = mapped_column(
		ForeignKey("operations.operation_id"), nullable=False
		)
		service_specification_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), nullable=False)
		qos_profile: Mapped[str] = mapped_column(String(256), nullable=False)
		duration_seconds: Mapped[int] = mapped_column(Integer, nullable=False)
		state: Mapped[QodSessionState] = mapped_column(_enum_type(QodSessionState), nullable=False)
		external_ref: Mapped[str \| None] = mapped_column(String(255))
		device_ports: Mapped[dict[str, object] \| None] = mapped_column(JSONB)
		application_server_ports: Mapped[dict[str, object] \| None] = mapped_column(JSONB)

src/open_exposure_gateway/adapters/databus/nats_adapter.py

+36 −0

Original line number	Diff line number	Diff line
		@@ -10,6 +10,7 @@ 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.quality_on_demand import SRMOperationStatus
		from open_exposure_gateway.ports.databus_port import DataBusPort

		logger: structlog.BoundLogger = structlog.get_logger(__name__)
		@@ -109,3 +110,38 @@ class NatsOperationConsumer:
		await self._handler(event)
		except Exception:
		logger.exception("operation_completed_handler_failed", operation_id=event.operation_id)


		class NatsOperationStatusConsumer:
		def __init__(
		self,
		client: Client,
		subject: str,
		handler: Callable[[SRMOperationStatus], Awaitable[None]],
		) -> None:
		self._client = client
		self._subject = subject
		self._handler = handler
		self._subscription: Subscription \| None = None

		async def start(self) -> None:
		# TODO: same at-most-once caveat as NatsOperationConsumer.start() — see there.
		self._subscription = await self._client.subscribe(self._subject, cb=self._handle_message)

		async def _handle_message(self, msg: _Msg) -> None:
		try:
		raw: Any = json.loads(msg.data.decode())
		except (json.JSONDecodeError, UnicodeDecodeError):
		logger.warning("invalid_json", subject=msg.subject)
		return

		try:
		event = SRMOperationStatus.model_validate(raw)
		except ValidationError as exc:
		logger.warning("invalid_operation_status_event", subject=msg.subject, error=str(exc))
		return

		try:
		await self._handler(event)
		except Exception:
		logger.exception("operation_status_handler_failed", operation_id=event.operation_id)