feat(eam): handle event.srm.operation.completed, update operations row (e51075f3) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/databus/nats_adapter.py

+20 −4

Original line number	Diff line number	Diff line
		import json
		from collections.abc import Awaitable, Callable
		from typing import Any, Protocol

		import nats
		import structlog
		from nats.aio.client import Client
		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.ports.databus_port import DataBusPort

		logger: structlog.BoundLogger = structlog.get_logger(__name__)
		@@ -71,9 +74,15 @@ class _Msg(Protocol):


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

		async def start(self) -> None:
		@@ -90,6 +99,13 @@ class NatsOperationConsumer:
		logger.warning("invalid_json", subject=msg.subject)
		return

		# TODO: parse operation.completed payload, update oeg_db operation record to
		# COMPLETED or FAILED, and trigger webhook callback if registered
		logger.info("operation_completed_received", subject=msg.subject, payload=raw)
		try:
		event = SRMOperationCompleted.model_validate(raw)
		except ValidationError as exc:
		logger.warning("invalid_operation_completed_event", subject=msg.subject, error=str(exc))
		return

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

src/open_exposure_gateway/application/services/edge_application_management_service.py

+45 −1

Original line number	Diff line number	Diff line
		from datetime import datetime, timezone
		from typing import Optional
		from typing import Any, Optional
		from uuid import UUID, uuid4

		import structlog
		@@ -36,6 +36,7 @@ from open_exposure_gateway.core.exceptions import (
		from open_exposure_gateway.domain.edge_application_management import (
		ResourceZone,
		SRMCatalogPayload,
		SRMOperationCompleted,
		Subject,
		)
		from open_exposure_gateway.domain.models import (
		@@ -64,6 +65,12 @@ logger: structlog.BoundLogger = structlog.get_logger(__name__)
		# catalog or left to fail at deploy time (ADR-0009).
		_SUPPORTED_PACKAGE_TYPES = frozenset({"CONTAINER", "HELM"})

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


		class EdgeApplicationManagementService:
		def __init__(
		@@ -347,3 +354,40 @@ class EdgeApplicationManagementService:
		command,
		"Failed to publish app instance termination command",
		)

		async def handle_completed(self, event: SRMOperationCompleted) -> None:
		if self._operation_repo is None:
		raise RuntimeError("OperationRepository is 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

		status = _OPERATION_COMPLETION_STATUS_MAP[event.status]
		result: Optional[dict[str, Any]] = None
		if status != OperationStatus.FAILED:
		result = {
		"instances": [
		{
		"app_instance_id": instance.service_instance_id,
		"service_instance_id": instance.service_instance_id,
		"zone_id": instance.zone_id,
		"status": instance.status,
		}
		for instance in event.instances
		]
		}

		updated = operation.model_copy(
		update={
		"status": status,
		"result": result,
		"error": event.error,
		"completed_at": datetime.fromisoformat(event.completed_at),
		}
		)
		await self._operation_repo.save(updated)

src/open_exposure_gateway/main.py

+32 −1

Original line number	Diff line number	Diff line
		@@ -4,12 +4,14 @@ from typing import Optional

		import structlog
		from fastapi import FastAPI, Request, Response
		from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
		from starlette.types import Lifespan

		from open_exposure_gateway.adapters.database.core import (
		build_engine_and_session_maker,
		schema_initialization,
		)
		from open_exposure_gateway.adapters.database.repos.operations import SqlOperationRepository
		from open_exposure_gateway.adapters.databus.nats_adapter import (
		NatsMessagePublisher,
		NatsOperationConsumer,
		@@ -26,9 +28,37 @@ from open_exposure_gateway.api.error_handlers import (
		x_correlator_header,
		)
		from open_exposure_gateway.api.platform.health import router as health_router
		from open_exposure_gateway.application.services.edge_application_management_service import (
		EdgeApplicationManagementService,
		)
		from open_exposure_gateway.core.config import get_settings
		from open_exposure_gateway.core.logging import configure_logging
		from open_exposure_gateway.domain.edge_application_management import Subject
		from open_exposure_gateway.domain.edge_application_management import (
		SRMOperationCompleted,
		Subject,
		)
		from open_exposure_gateway.ports.srm_port import SRMClientPort


		def _build_operation_completed_handler(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		) -> Callable[[SRMOperationCompleted], Awaitable[None]]:
		async def handle(event: SRMOperationCompleted) -> None:
		async with session_maker() as session:
		try:
		service = EdgeApplicationManagementService(
		srm_client=srm_client,
		operation_repo=SqlOperationRepository(session),
		)
		await service.handle_completed(event)
		await session.commit()
		except Exception:
		await session.rollback()
		raise

		return handle


		openapi_tags = [
		{
		@@ -94,6 +124,7 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		consumer = NatsOperationConsumer(
		client=publisher.client,
		subject=Subject.OPERATION_COMPLETED,
		handler=_build_operation_completed_handler(session_maker, srm_client),
		)
		await consumer.start()
		logger.info("NATS consumer started", subject=Subject.OPERATION_COMPLETED)

tests/conformance/conftest.py

+3 −2

Original line number	Diff line number	Diff line
		@@ -51,13 +51,14 @@ def service_overrides() -> Generator[None, None, None]:
		provider="conformance-provider",
		)
		)
		wire_operation_consumer(bus)
		operation_repo = FakeOperationRepository()
		wire_operation_consumer(bus, operation_repo)
		wire_srm_worker(bus, srm)
		app.dependency_overrides[get_edge_app_service] = lambda: EdgeApplicationManagementService(
		srm_client=srm,
		publisher=bus,
		app_registration_repo=FakeAppRegistrationRepository(),
		operation_repo=FakeOperationRepository(),
		operation_repo=operation_repo,
		app_instance_repo=FakeAppInstanceRepository(),
		callback_registration_repo=FakeCallbackRegistrationRepository(),
		)

tests/integration/test_nats_consumer.py

+16 −5

Original line number	Diff line number	Diff line
		import asyncio
		import json
		from typing import Any
		from unittest.mock import AsyncMock

		import nats
		from nats.aio.client import Client
		@@ -9,14 +10,18 @@ from open_exposure_gateway.adapters.databus.nats_adapter import NatsOperationCon


		async def test_consumer_subscribes_to_subject(nats_client: Client) -> None:
		consumer = NatsOperationConsumer(client=nats_client, subject="operation.completed")
		consumer = NatsOperationConsumer(
		client=nats_client, subject="operation.completed", handler=AsyncMock()
		)
		await consumer.start()
		assert consumer._subscription is not None


		async def test_consumer_invokes_handler_when_message_arrives(nats_client: Client) -> None:
		invoked = asyncio.Event()
		consumer = NatsOperationConsumer(client=nats_client, subject="operation.completed")
		consumer = NatsOperationConsumer(
		client=nats_client, subject="operation.completed", handler=AsyncMock()
		)

		original = consumer._handle_message

		@@ -34,7 +39,9 @@ async def test_consumer_invokes_handler_when_message_arrives(nats_client: Client
		async def test_consumer_decodes_json_payload(nats_client: Client) -> None:
		decoded: list[Any] = []
		ready = asyncio.Event()
		consumer = NatsOperationConsumer(client=nats_client, subject="operation.completed")
		consumer = NatsOperationConsumer(
		client=nats_client, subject="operation.completed", handler=AsyncMock()
		)

		original = consumer._handle_message

		@@ -55,7 +62,9 @@ async def test_consumer_decodes_json_payload(nats_client: Client) -> None:

		async def test_consumer_handles_invalid_json_gracefully(nats_client: Client) -> None:
		handled = asyncio.Event()
		consumer = NatsOperationConsumer(client=nats_client, subject="operation.completed")
		consumer = NatsOperationConsumer(
		client=nats_client, subject="operation.completed", handler=AsyncMock()
		)

		original = consumer._handle_message

		@@ -73,7 +82,9 @@ async def test_consumer_handles_invalid_json_gracefully(nats_client: Client) ->
		async def test_consumer_unsubscribes_cleanly(nats_url: str) -> None:
		client = await nats.connect(nats_url)
		try:
		consumer = NatsOperationConsumer(client=client, subject="operation.completed")
		consumer = NatsOperationConsumer(
		client=client, subject="operation.completed", handler=AsyncMock()
		)
		await consumer.start()
		assert consumer._subscription is not None

Admin message