Merge branch 'refactor/app-deployment' into 'refactor/tf-sdk-ports' (d10e87df) · Commits · OOP / code / service-resource-manager

src/srm/adapters/database/repos/catalog.py

+20 −0

Original line number	Diff line number	Diff line
		@@ -147,6 +147,16 @@ class SqlServiceDeploymentUnitRepository(ServiceDeploymentUnitRepository):
		row = await self._session.scalar(stmt)
		return ServiceDeploymentUnitMapper.to_domain(row) if row is not None else None

		async def list_by_service_specification_id(
		self,
		service_specification_id: UUID,
		) -> list[ServiceDeploymentUnit]:
		stmt = select(ServiceDeploymentUnitRow).where(
		ServiceDeploymentUnitRow.service_specification_id == service_specification_id
		)
		rows = (await self._session.scalars(stmt)).all()
		return [ServiceDeploymentUnitMapper.to_domain(row) for row in rows]

		async def create(self, deployment_unit: ServiceDeploymentUnit) -> ServiceDeploymentUnit:
		row = ServiceDeploymentUnitMapper.to_row(deployment_unit)
		self._session.add(row)
		@@ -173,6 +183,16 @@ class SqlServiceCapabilityRequirementRepository(ServiceCapabilityRequirementRepo
		row = await self._session.scalar(stmt)
		return ServiceCapabilityRequirementMapper.to_domain(row) if row is not None else None

		async def list_by_service_specification_id(
		self,
		service_specification_id: UUID,
		) -> list[ServiceCapabilityRequirement]:
		stmt = select(ServiceCapabilityRequirementRow).where(
		ServiceCapabilityRequirementRow.service_specification_id == service_specification_id
		)
		rows = (await self._session.scalars(stmt)).all()
		return [ServiceCapabilityRequirementMapper.to_domain(row) for row in rows]

		async def create(
		self,
		capability_requirement: ServiceCapabilityRequirement,

src/srm/adapters/database/repos/runtime_inventory.py

+56 −0

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

		from datetime import datetime
		from uuid import UUID

		import structlog
		@@ -19,6 +20,7 @@ from srm.domain.models.runtime_inventory import (
		ServiceInstance,
		ServiceOrder,
		)
		from srm.domain.models.runtime_inventory.enums import ServiceInstanceState, ServiceOrderState
		from srm.domain.ports.database.runtime_inventory import (
		CapabilityInstanceRepository,
		ServiceInstanceRepository,
		@@ -51,6 +53,11 @@ class SqlServiceOrderRepository(ServiceOrderRepository):
		row = await self._session.scalar(stmt)
		return ServiceOrderMapper.to_domain(row) if row is not None else None

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

		async def create(self, service_order: ServiceOrder) -> ServiceOrder:
		row = ServiceOrderMapper.to_row(service_order)
		self._session.add(row)
		@@ -65,6 +72,26 @@ class SqlServiceOrderRepository(ServiceOrderRepository):
		raise RuntimeError("Created service order could not be reloaded")
		return saved

		async def update_state(
		self,
		id: UUID,
		state: ServiceOrderState,
		*,
		completed_at: datetime \| None = None,
		failure_detail: dict[str, object] \| None = None,
		) -> ServiceOrder:
		row = await self._session.get(ServiceOrderRow, id)
		if row is None:
		raise RuntimeError("Service order could not be reloaded")
		row.state = state
		row.completed_at = completed_at
		row.failure_detail = failure_detail
		await self._session.flush()
		saved = await self.get_by_id(id)
		if saved is None:
		raise RuntimeError("Updated service order could not be reloaded")
		return saved


		class SqlServiceInstanceRepository(ServiceInstanceRepository):
		def __init__(self, session: AsyncSession) -> None:
		@@ -81,6 +108,20 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository):
		row = await self._session.scalar(stmt)
		return ServiceInstanceMapper.to_domain(row) if row is not None else None

		async def list_by_originating_service_order_id(
		self,
		originating_service_order_id: UUID,
		) -> list[ServiceInstance]:
		stmt = (
		select(ServiceInstanceRow)
		.options(
		selectinload(ServiceInstanceRow.capability_instances),
		)
		.where(ServiceInstanceRow.originating_service_order_id == originating_service_order_id)
		)
		rows = (await self._session.scalars(stmt)).all()
		return [ServiceInstanceMapper.to_domain(row) for row in rows]

		async def create(self, service_instance: ServiceInstance) -> ServiceInstance:
		row = ServiceInstanceMapper.to_row(service_instance)
		self._session.add(row)
		@@ -95,6 +136,21 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository):
		raise RuntimeError("Created service instance could not be reloaded")
		return saved

		async def update_state(
		self,
		id: UUID,
		state: ServiceInstanceState,
		) -> ServiceInstance:
		row = await self._session.get(ServiceInstanceRow, id)
		if row is None:
		raise RuntimeError("Service instance could not be reloaded")
		row.state = state
		await self._session.flush()
		saved = await self.get_by_id(id)
		if saved is None:
		raise RuntimeError("Updated service instance could not be reloaded")
		return saved


		class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):
		def __init__(self, session: AsyncSession) -> None:

src/srm/adapters/database/repos/topology.py

+15 −0

Original line number	Diff line number	Diff line
		@@ -16,6 +16,7 @@ from srm.adapters.database.mappers import (
		)
		from srm.adapters.database.sql import CapabilityRow, ControlPathBindingRow, DomainRow, ZoneRow
		from srm.domain.models.topology import Capability, ControlPathBinding, Domain, Zone
		from srm.domain.models.topology.enums import ZoneKind, ZoneState
		from srm.domain.ports.database.topology import (
		CapabilityRepository,
		ControlPathBindingRepository,
		@@ -57,6 +58,20 @@ class SqlZoneRepository(ZoneRepository):
		row = await self._session.scalar(stmt)
		return ZoneMapper.to_domain(row) if row is not None else None

		async def list_active_resource_zones(self) -> list[Zone]:
		stmt = (
		select(ZoneRow)
		.options(
		selectinload(ZoneRow.domains)
		.selectinload(DomainRow.capabilities)
		.selectinload(CapabilityRow.control_path_bindings)
		)
		.where(ZoneRow.kind == ZoneKind.RESOURCE, ZoneRow.state == ZoneState.ACTIVE)
		.order_by(ZoneRow.ref, ZoneRow.id)
		)
		rows = (await self._session.scalars(stmt)).all()
		return [ZoneMapper.to_domain(row) for row in rows]

		async def create(self, zone: Zone) -> Zone:
		row = ZoneMapper.to_row(zone)
		self._session.add(row)

src/srm/api/databus/dependencies.py

0 → 100644

+53 −0

Original line number	Diff line number	Diff line
		from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

		from srm.adapters.database.repos.catalog import (
		SqlServiceCapabilityRequirementRepository,
		SqlServiceDeploymentUnitRepository,
		SqlServiceSpecificationRepository,
		)
		from srm.adapters.database.repos.runtime_inventory import (
		SqlCapabilityInstanceRepository,
		SqlServiceInstanceRepository,
		SqlServiceOrderRepository,
		)
		from srm.adapters.database.repos.topology import SqlDomainRepository, 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.use_cases.deploy_service import DeployServiceUseCase
		from srm.domain.ports.databus.publisher import DataBusPublisher


		def get_deploy_service_use_case(
		session: AsyncSession,
		connection_manager: NatsConnectionManager,
		) -> DeployServiceUseCase:
		return build_deploy_service_use_case(session, NatsPublisher(connection_manager))


		def build_deploy_service_use_case(
		session: AsyncSession,
		publisher: DataBusPublisher,
		) -> DeployServiceUseCase:
		return DeployServiceUseCase(
		service_specifications=SqlServiceSpecificationRepository(session),
		service_deployment_units=SqlServiceDeploymentUnitRepository(session),
		service_capability_requirements=SqlServiceCapabilityRequirementRepository(session),
		service_orders=SqlServiceOrderRepository(session),
		service_instances=SqlServiceInstanceRepository(session),
		capability_instances=SqlCapabilityInstanceRepository(session),
		zones=SqlZoneRepository(session),
		domains=SqlDomainRepository(session),
		publisher=publisher,
		)


		def get_deploy_service_command_coordinator(
		session_maker: async_sessionmaker[AsyncSession],
		connection_manager: NatsConnectionManager,
		) -> DeployServiceCommandCoordinator:
		return DeployServiceCommandCoordinator(
		session_maker=session_maker,
		use_case_factory=build_deploy_service_use_case,
		publisher=NatsPublisher(connection_manager),
		)

src/srm/api/databus/nats_subscriber.py

+51 −18

Original line number	Diff line number	Diff line
		@@ -3,34 +3,59 @@ from collections.abc import Awaitable, Callable
		import structlog
		from nats.aio.msg import Msg
		from nats.aio.subscription import Subscription
		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.schemas import InboundMessage
		from srm.api.databus.dependencies import get_deploy_service_command_coordinator
		from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage

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

		COMMAND_SUBJECTS = [
		"command.srm.service.deploy",
		"command.srm.service.scale",
		"command.srm.service.terminate",
		"command.srm.network.capability.activate",
		"command.srm.network.capability.update",
		"command.srm.network.capability.deactivate",
		]

		MessageHandler = Callable[[AnswerableCommandV1], Awaitable[None]]


		async def _noop_callback(envelope: AnswerableCommandV1) -> None:
		# TODO: non-deploy command handlers must validate in two stages (interface-contract.md §A).
		logger.info(
		"Message Received",
		operation_id=str(envelope.operation_id),
		correlation_id=envelope.correlation_id,
		)


		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,
		}

		async def _noop_router(message: InboundMessage) -> None:
		# TODO: the real router must validate in two stages (interface-contract.md §A, ADR-0032).
		logger.info("Message Received", subject=message.subject, data=message.payload)
		return None
		COMMAND_SUBJECTS = ("command.srm.service.deploy", *COMMAND_HANDLERS.keys())


		async def subscribe_to_subjects(
		connection_manager: NatsConnectionManager,
		session_maker: async_sessionmaker[AsyncSession],
		) -> list["NatsSubscriber"]:
		deploy_coordinator = get_deploy_service_command_coordinator(
		session_maker,
		connection_manager,
		)

		command_handlers = {
		"command.srm.service.deploy": deploy_coordinator.handle,
		**COMMAND_HANDLERS,
		}
		subscribers = [
		NatsSubscriber(connection_manager=connection_manager, subject=subject, router=_noop_router)
		for subject in COMMAND_SUBJECTS
		NatsSubscriber(
		connection_manager=connection_manager,
		subject=subject,
		callback=callback,
		)
		for subject, callback in command_handlers.items()
		]

		for subscriber in subscribers:
		@@ -44,11 +69,11 @@ class NatsSubscriber:
		self,
		connection_manager: NatsConnectionManager,
		subject: str,
		router: Callable[[InboundMessage], Awaitable[None]],
		callback: MessageHandler,
		) -> None:
		self._connection_manager = connection_manager
		self._subject = subject
		self._router = router
		self._callback = callback
		self._subscription: Subscription \| None = None

		async def start(self) -> None:
		@@ -64,6 +89,14 @@ class NatsSubscriber:
		headers=dict(msg.headers) if msg.headers is not None else {},
		)
		try:
		await self._router(inbound_message)
		envelope = AnswerableCommandV1.model_validate_json(inbound_message.payload)
		except ValidationError:
		# Stage 1 failures are unanswerable: publish no SRM event;
		# route to JetStream/DLQ when wired.
		logger.exception("databus_stage1_unanswerable", subject=msg.subject)
		return

		try:
		await self._callback(envelope)
		except Exception:
		logger.exception("databus_handler_failed", subject=msg.subject)