Loading src/srm/adapters/database/repos/runtime_inventory.py +28 −1 Changes for src/srm/adapters/database/repos/runtime_inventory.py: 28 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -20,7 +20,11 @@ from srm.domain.models.runtime_inventory import ( ServiceInstance, ServiceOrder, ) from srm.domain.models.runtime_inventory.enums import ServiceInstanceState, ServiceOrderState from srm.domain.models.runtime_inventory.enums import ( CapabilityInstanceState, ServiceInstanceState, ServiceOrderState, ) from srm.domain.ports.database.runtime_inventory import ( CapabilityInstanceRepository, ServiceInstanceRepository, Loading Loading @@ -140,11 +144,15 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository): self, id: UUID, state: ServiceInstanceState, *, terminated_at: datetime | None = None, ) -> ServiceInstance: row = await self._session.get(ServiceInstanceRow, id) if row is None: raise RuntimeError("Service instance could not be reloaded") row.state = state if terminated_at is not None: row.terminated_at = terminated_at await self._session.flush() saved = await self.get_by_id(id) if saved is None: Loading Loading @@ -174,3 +182,22 @@ class SqlCapabilityInstanceRepository(CapabilityInstanceRepository): if saved is None: raise RuntimeError("Created capability instance could not be reloaded") return saved async def update_state( self, id: UUID, state: CapabilityInstanceState, *, terminated_at: datetime | None = None, ) -> CapabilityInstance: row = await self._session.get(CapabilityInstanceRow, id) if row is None: raise RuntimeError("Capability instance could not be reloaded") row.state = state if terminated_at is not None: row.terminated_at = terminated_at await self._session.flush() saved = await self.get_by_id(id) if saved is None: raise RuntimeError("Updated capability instance could not be reloaded") return saved src/srm/api/databus/dependencies.py +32 −0 Changes for src/srm/api/databus/dependencies.py: 32 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -14,7 +14,9 @@ from srm.adapters.database.repos.topology import SqlZoneRepository from srm.adapters.databus.nats_connection_manager import NatsConnectionManager from srm.adapters.databus.nats_publisher import NatsPublisher from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator from srm.application.command_handlers.terminate_service import TerminateServiceCommandCoordinator from srm.application.use_cases.deploy_service import DeployServiceUseCase from srm.application.use_cases.terminate_service import TerminateServiceUseCase from srm.domain.ports.databus.publisher import DataBusPublisher Loading Loading @@ -50,3 +52,33 @@ def get_deploy_service_command_coordinator( use_case_factory=build_deploy_service_use_case, publisher=NatsPublisher(connection_manager), ) def get_terminate_service_use_case( session: AsyncSession, connection_manager: NatsConnectionManager, ) -> TerminateServiceUseCase: return build_terminate_service_use_case(session, NatsPublisher(connection_manager)) def build_terminate_service_use_case( session: AsyncSession, publisher: DataBusPublisher, ) -> TerminateServiceUseCase: return TerminateServiceUseCase( service_orders=SqlServiceOrderRepository(session), service_instances=SqlServiceInstanceRepository(session), capability_instances=SqlCapabilityInstanceRepository(session), publisher=publisher, ) def get_terminate_service_command_coordinator( session_maker: async_sessionmaker[AsyncSession], connection_manager: NatsConnectionManager, ) -> TerminateServiceCommandCoordinator: return TerminateServiceCommandCoordinator( session_maker=session_maker, use_case_factory=build_terminate_service_use_case, publisher=NatsPublisher(connection_manager), ) src/srm/api/databus/nats_subscriber.py +14 −3 Changes for src/srm/api/databus/nats_subscriber.py: 14 added lines, 3 removed lines. Original line number Diff line number Diff line Loading @@ -7,7 +7,10 @@ from pydantic import ValidationError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from srm.adapters.databus.nats_connection_manager import NatsConnectionManager from srm.api.databus.dependencies import get_deploy_service_command_coordinator from srm.api.databus.dependencies import ( get_deploy_service_command_coordinator, get_terminate_service_command_coordinator, ) from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage logger: structlog.BoundLogger = structlog.get_logger(__name__) Loading @@ -27,13 +30,16 @@ async def _noop_callback(envelope: AnswerableCommandV1) -> None: COMMAND_HANDLERS: dict[str, MessageHandler] = { "command.srm.service.scale": _noop_callback, "command.srm.service.terminate": _noop_callback, "command.srm.network.capability.activate": _noop_callback, "command.srm.network.capability.update": _noop_callback, "command.srm.network.capability.deactivate": _noop_callback, } COMMAND_SUBJECTS = ("command.srm.service.deploy", *COMMAND_HANDLERS.keys()) COMMAND_SUBJECTS = ( "command.srm.service.deploy", "command.srm.service.terminate", *COMMAND_HANDLERS.keys(), ) async def subscribe_to_subjects( Loading @@ -44,9 +50,14 @@ async def subscribe_to_subjects( session_maker, connection_manager, ) terminate_coordinator = get_terminate_service_command_coordinator( session_maker, connection_manager, ) command_handlers = { "command.srm.service.deploy": deploy_coordinator.handle, "command.srm.service.terminate": terminate_coordinator.handle, **COMMAND_HANDLERS, } subscribers = [ Loading src/srm/application/command_handlers/terminate_service.py 0 → 100644 +227 −0 Changes for src/srm/application/command_handlers/terminate_service.py: 227 added lines, 0 removed lines. Original line number Diff line number Diff line from __future__ import annotations from collections.abc import Callable from datetime import UTC, datetime from uuid import UUID from pydantic import ValidationError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from srm.api.databus.schemas import AnswerableCommandV1, SrmServiceTerminateV1 from srm.application.use_cases.terminate_service import ( TerminatePayloadCommand, TerminateServiceCommand, TerminateServiceResult, TerminateServiceUseCase, ) from srm.domain.ports.databus.events import ( OperationCompletedStatus, OperationStatusState, SrmOperationCompletedV1, SrmOperationStatusV1, ) from srm.domain.ports.databus.publisher import DataBusPublisher TerminateUseCaseFactory = Callable[[AsyncSession, DataBusPublisher], TerminateServiceUseCase] class TerminateServiceCommandCoordinator: def __init__( self, session_maker: async_sessionmaker[AsyncSession], use_case_factory: TerminateUseCaseFactory, publisher: DataBusPublisher, ) -> None: self._session_maker = session_maker self._use_case_factory = use_case_factory self._publisher = publisher async def handle(self, envelope: AnswerableCommandV1) -> None: command = await self._parse_command(envelope) if command is None: return accepted = await self._accept(command) if accepted.completed_event is not None: await self._publisher.publish( "event.srm.operation.completed", accepted.completed_event.model_dump(mode="json"), ) return if accepted.duplicate_order_id is not None: await self._publish_accepted(envelope, accepted.duplicate_order_id) return if accepted.accepted_order_id is None: return await self._publish_accepted(envelope, accepted.accepted_order_id) try: completed_event = await self._complete(command, accepted) except Exception as exc: completed_event = await self._fail_after_accept(command, accepted, exc) if completed_event is not None: await self._publisher.publish( "event.srm.operation.completed", completed_event.model_dump(mode="json"), ) async def _parse_command( self, envelope: AnswerableCommandV1, ) -> TerminateServiceCommand | None: try: terminate_message = SrmServiceTerminateV1.model_validate( envelope.model_dump(mode="python") ) except ValidationError: await self._publish_failed_before_start_for_invalid_command(envelope) return None return TerminateServiceCommand( operation_id=terminate_message.operation_id, correlation_id=terminate_message.correlation_id, requested_at=terminate_message.requested_at, app_provider_id=terminate_message.app_provider_id, federation_partner_ref=terminate_message.federation_partner_ref, source=terminate_message.source, service_instance_id=terminate_message.service_instance_id, service_specification_id=terminate_message.service_specification_id, terminate=TerminatePayloadCommand( grace_period_seconds=terminate_message.terminate.grace_period_seconds, ), payload_snapshot=terminate_message.model_dump(mode="json"), ) async def _accept(self, command: TerminateServiceCommand) -> TerminateServiceResult: async with self._session_maker() as session: try: use_case = self._use_case_factory(session, self._publisher) accepted = await use_case.accept(command) await session.commit() return accepted except Exception: await session.rollback() raise async def _complete( self, command: TerminateServiceCommand, accepted: TerminateServiceResult, ) -> SrmOperationCompletedV1 | None: async with self._session_maker() as session: try: use_case = self._use_case_factory(session, self._publisher) completed_event = await use_case.complete(command, accepted) await session.commit() return completed_event except Exception: await session.rollback() raise async def _fail_after_accept( self, command: TerminateServiceCommand, accepted: TerminateServiceResult, error: Exception, ) -> SrmOperationCompletedV1: async with self._session_maker() as session: try: use_case = self._use_case_factory(session, self._publisher) completed_event = await use_case.fail_after_accept(command, accepted, error) await session.commit() return completed_event except Exception: await session.rollback() return self._failed_after_accept_event(command, accepted, error) async def _publish_accepted( self, envelope: AnswerableCommandV1, service_order_id: UUID, ) -> None: model_extra = envelope.model_extra or {} status_event = SrmOperationStatusV1( schema_version="1.0", operation_id=envelope.operation_id, service_order_id=service_order_id, service_instance_id=model_extra.get("service_instance_id"), capability="terminate", state=OperationStatusState.ACCEPTED, correlation_id=envelope.correlation_id, emitted_at=datetime.now(UTC), ) await self._publisher.publish( "event.srm.operation.status", status_event.model_dump(mode="json"), ) async def _publish_failed_before_start_for_invalid_command( self, envelope: AnswerableCommandV1, ) -> None: now = datetime.now(UTC) status_event = SrmOperationStatusV1( schema_version="1.0", operation_id=envelope.operation_id, service_order_id=None, service_instance_id=None, capability="terminate", state=OperationStatusState.FAILED_BEFORE_START, correlation_id=envelope.correlation_id, emitted_at=now, ) completed_event = SrmOperationCompletedV1( schema_version="1.0", operation_id=envelope.operation_id, status=OperationCompletedStatus.FAILED, service_order_id=None, error={ "type": "about:blank", "title": "Terminate service command is invalid.", "status": 400, }, correlation_id=envelope.correlation_id, completed_at=now, ) await self._publisher.publish( "event.srm.operation.status", status_event.model_dump(mode="json"), ) await self._publisher.publish( "event.srm.operation.completed", completed_event.model_dump(mode="json"), ) def _failed_after_accept_event( self, command: TerminateServiceCommand, accepted: TerminateServiceResult, error: Exception, ) -> SrmOperationCompletedV1: now = datetime.now(UTC) problem = { "type": "about:blank", "title": "Terminate service failed after acceptance.", "status": 500, "detail": str(error), } return SrmOperationCompletedV1( schema_version="1.0", operation_id=command.operation_id, status=OperationCompletedStatus.FAILED, service_order_id=accepted.accepted_order_id, instances=[ { "service_instance_id": str(command.service_instance_id), "zone_id": str(accepted.target_zone_id), "status": "failed", "error": problem, } ] if accepted.target_zone_id is not None else [], error=None if accepted.target_zone_id is not None else problem, correlation_id=command.correlation_id, completed_at=now, ) src/srm/application/use_cases/terminate_service.py 0 → 100644 +330 −0 File added.Preview size limit exceeded, changes collapsed. Show changes Loading
src/srm/adapters/database/repos/runtime_inventory.py +28 −1 Changes for src/srm/adapters/database/repos/runtime_inventory.py: 28 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -20,7 +20,11 @@ from srm.domain.models.runtime_inventory import ( ServiceInstance, ServiceOrder, ) from srm.domain.models.runtime_inventory.enums import ServiceInstanceState, ServiceOrderState from srm.domain.models.runtime_inventory.enums import ( CapabilityInstanceState, ServiceInstanceState, ServiceOrderState, ) from srm.domain.ports.database.runtime_inventory import ( CapabilityInstanceRepository, ServiceInstanceRepository, Loading Loading @@ -140,11 +144,15 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository): self, id: UUID, state: ServiceInstanceState, *, terminated_at: datetime | None = None, ) -> ServiceInstance: row = await self._session.get(ServiceInstanceRow, id) if row is None: raise RuntimeError("Service instance could not be reloaded") row.state = state if terminated_at is not None: row.terminated_at = terminated_at await self._session.flush() saved = await self.get_by_id(id) if saved is None: Loading Loading @@ -174,3 +182,22 @@ class SqlCapabilityInstanceRepository(CapabilityInstanceRepository): if saved is None: raise RuntimeError("Created capability instance could not be reloaded") return saved async def update_state( self, id: UUID, state: CapabilityInstanceState, *, terminated_at: datetime | None = None, ) -> CapabilityInstance: row = await self._session.get(CapabilityInstanceRow, id) if row is None: raise RuntimeError("Capability instance could not be reloaded") row.state = state if terminated_at is not None: row.terminated_at = terminated_at await self._session.flush() saved = await self.get_by_id(id) if saved is None: raise RuntimeError("Updated capability instance could not be reloaded") return saved
src/srm/api/databus/dependencies.py +32 −0 Changes for src/srm/api/databus/dependencies.py: 32 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -14,7 +14,9 @@ from srm.adapters.database.repos.topology import SqlZoneRepository from srm.adapters.databus.nats_connection_manager import NatsConnectionManager from srm.adapters.databus.nats_publisher import NatsPublisher from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator from srm.application.command_handlers.terminate_service import TerminateServiceCommandCoordinator from srm.application.use_cases.deploy_service import DeployServiceUseCase from srm.application.use_cases.terminate_service import TerminateServiceUseCase from srm.domain.ports.databus.publisher import DataBusPublisher Loading Loading @@ -50,3 +52,33 @@ def get_deploy_service_command_coordinator( use_case_factory=build_deploy_service_use_case, publisher=NatsPublisher(connection_manager), ) def get_terminate_service_use_case( session: AsyncSession, connection_manager: NatsConnectionManager, ) -> TerminateServiceUseCase: return build_terminate_service_use_case(session, NatsPublisher(connection_manager)) def build_terminate_service_use_case( session: AsyncSession, publisher: DataBusPublisher, ) -> TerminateServiceUseCase: return TerminateServiceUseCase( service_orders=SqlServiceOrderRepository(session), service_instances=SqlServiceInstanceRepository(session), capability_instances=SqlCapabilityInstanceRepository(session), publisher=publisher, ) def get_terminate_service_command_coordinator( session_maker: async_sessionmaker[AsyncSession], connection_manager: NatsConnectionManager, ) -> TerminateServiceCommandCoordinator: return TerminateServiceCommandCoordinator( session_maker=session_maker, use_case_factory=build_terminate_service_use_case, publisher=NatsPublisher(connection_manager), )
src/srm/api/databus/nats_subscriber.py +14 −3 Changes for src/srm/api/databus/nats_subscriber.py: 14 added lines, 3 removed lines. Original line number Diff line number Diff line Loading @@ -7,7 +7,10 @@ from pydantic import ValidationError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from srm.adapters.databus.nats_connection_manager import NatsConnectionManager from srm.api.databus.dependencies import get_deploy_service_command_coordinator from srm.api.databus.dependencies import ( get_deploy_service_command_coordinator, get_terminate_service_command_coordinator, ) from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage logger: structlog.BoundLogger = structlog.get_logger(__name__) Loading @@ -27,13 +30,16 @@ async def _noop_callback(envelope: AnswerableCommandV1) -> None: COMMAND_HANDLERS: dict[str, MessageHandler] = { "command.srm.service.scale": _noop_callback, "command.srm.service.terminate": _noop_callback, "command.srm.network.capability.activate": _noop_callback, "command.srm.network.capability.update": _noop_callback, "command.srm.network.capability.deactivate": _noop_callback, } COMMAND_SUBJECTS = ("command.srm.service.deploy", *COMMAND_HANDLERS.keys()) COMMAND_SUBJECTS = ( "command.srm.service.deploy", "command.srm.service.terminate", *COMMAND_HANDLERS.keys(), ) async def subscribe_to_subjects( Loading @@ -44,9 +50,14 @@ async def subscribe_to_subjects( session_maker, connection_manager, ) terminate_coordinator = get_terminate_service_command_coordinator( session_maker, connection_manager, ) command_handlers = { "command.srm.service.deploy": deploy_coordinator.handle, "command.srm.service.terminate": terminate_coordinator.handle, **COMMAND_HANDLERS, } subscribers = [ Loading
src/srm/application/command_handlers/terminate_service.py 0 → 100644 +227 −0 Changes for src/srm/application/command_handlers/terminate_service.py: 227 added lines, 0 removed lines. Original line number Diff line number Diff line from __future__ import annotations from collections.abc import Callable from datetime import UTC, datetime from uuid import UUID from pydantic import ValidationError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from srm.api.databus.schemas import AnswerableCommandV1, SrmServiceTerminateV1 from srm.application.use_cases.terminate_service import ( TerminatePayloadCommand, TerminateServiceCommand, TerminateServiceResult, TerminateServiceUseCase, ) from srm.domain.ports.databus.events import ( OperationCompletedStatus, OperationStatusState, SrmOperationCompletedV1, SrmOperationStatusV1, ) from srm.domain.ports.databus.publisher import DataBusPublisher TerminateUseCaseFactory = Callable[[AsyncSession, DataBusPublisher], TerminateServiceUseCase] class TerminateServiceCommandCoordinator: def __init__( self, session_maker: async_sessionmaker[AsyncSession], use_case_factory: TerminateUseCaseFactory, publisher: DataBusPublisher, ) -> None: self._session_maker = session_maker self._use_case_factory = use_case_factory self._publisher = publisher async def handle(self, envelope: AnswerableCommandV1) -> None: command = await self._parse_command(envelope) if command is None: return accepted = await self._accept(command) if accepted.completed_event is not None: await self._publisher.publish( "event.srm.operation.completed", accepted.completed_event.model_dump(mode="json"), ) return if accepted.duplicate_order_id is not None: await self._publish_accepted(envelope, accepted.duplicate_order_id) return if accepted.accepted_order_id is None: return await self._publish_accepted(envelope, accepted.accepted_order_id) try: completed_event = await self._complete(command, accepted) except Exception as exc: completed_event = await self._fail_after_accept(command, accepted, exc) if completed_event is not None: await self._publisher.publish( "event.srm.operation.completed", completed_event.model_dump(mode="json"), ) async def _parse_command( self, envelope: AnswerableCommandV1, ) -> TerminateServiceCommand | None: try: terminate_message = SrmServiceTerminateV1.model_validate( envelope.model_dump(mode="python") ) except ValidationError: await self._publish_failed_before_start_for_invalid_command(envelope) return None return TerminateServiceCommand( operation_id=terminate_message.operation_id, correlation_id=terminate_message.correlation_id, requested_at=terminate_message.requested_at, app_provider_id=terminate_message.app_provider_id, federation_partner_ref=terminate_message.federation_partner_ref, source=terminate_message.source, service_instance_id=terminate_message.service_instance_id, service_specification_id=terminate_message.service_specification_id, terminate=TerminatePayloadCommand( grace_period_seconds=terminate_message.terminate.grace_period_seconds, ), payload_snapshot=terminate_message.model_dump(mode="json"), ) async def _accept(self, command: TerminateServiceCommand) -> TerminateServiceResult: async with self._session_maker() as session: try: use_case = self._use_case_factory(session, self._publisher) accepted = await use_case.accept(command) await session.commit() return accepted except Exception: await session.rollback() raise async def _complete( self, command: TerminateServiceCommand, accepted: TerminateServiceResult, ) -> SrmOperationCompletedV1 | None: async with self._session_maker() as session: try: use_case = self._use_case_factory(session, self._publisher) completed_event = await use_case.complete(command, accepted) await session.commit() return completed_event except Exception: await session.rollback() raise async def _fail_after_accept( self, command: TerminateServiceCommand, accepted: TerminateServiceResult, error: Exception, ) -> SrmOperationCompletedV1: async with self._session_maker() as session: try: use_case = self._use_case_factory(session, self._publisher) completed_event = await use_case.fail_after_accept(command, accepted, error) await session.commit() return completed_event except Exception: await session.rollback() return self._failed_after_accept_event(command, accepted, error) async def _publish_accepted( self, envelope: AnswerableCommandV1, service_order_id: UUID, ) -> None: model_extra = envelope.model_extra or {} status_event = SrmOperationStatusV1( schema_version="1.0", operation_id=envelope.operation_id, service_order_id=service_order_id, service_instance_id=model_extra.get("service_instance_id"), capability="terminate", state=OperationStatusState.ACCEPTED, correlation_id=envelope.correlation_id, emitted_at=datetime.now(UTC), ) await self._publisher.publish( "event.srm.operation.status", status_event.model_dump(mode="json"), ) async def _publish_failed_before_start_for_invalid_command( self, envelope: AnswerableCommandV1, ) -> None: now = datetime.now(UTC) status_event = SrmOperationStatusV1( schema_version="1.0", operation_id=envelope.operation_id, service_order_id=None, service_instance_id=None, capability="terminate", state=OperationStatusState.FAILED_BEFORE_START, correlation_id=envelope.correlation_id, emitted_at=now, ) completed_event = SrmOperationCompletedV1( schema_version="1.0", operation_id=envelope.operation_id, status=OperationCompletedStatus.FAILED, service_order_id=None, error={ "type": "about:blank", "title": "Terminate service command is invalid.", "status": 400, }, correlation_id=envelope.correlation_id, completed_at=now, ) await self._publisher.publish( "event.srm.operation.status", status_event.model_dump(mode="json"), ) await self._publisher.publish( "event.srm.operation.completed", completed_event.model_dump(mode="json"), ) def _failed_after_accept_event( self, command: TerminateServiceCommand, accepted: TerminateServiceResult, error: Exception, ) -> SrmOperationCompletedV1: now = datetime.now(UTC) problem = { "type": "about:blank", "title": "Terminate service failed after acceptance.", "status": 500, "detail": str(error), } return SrmOperationCompletedV1( schema_version="1.0", operation_id=command.operation_id, status=OperationCompletedStatus.FAILED, service_order_id=accepted.accepted_order_id, instances=[ { "service_instance_id": str(command.service_instance_id), "zone_id": str(accepted.target_zone_id), "status": "failed", "error": problem, } ] if accepted.target_zone_id is not None else [], error=None if accepted.target_zone_id is not None else problem, correlation_id=command.correlation_id, completed_at=now, )
src/srm/application/use_cases/terminate_service.py 0 → 100644 +330 −0 File added.Preview size limit exceeded, changes collapsed. Show changes
