SRM Refactor: Federation Manager Client ports and adapters (!28) · Merge requests · OOP / code / service-resource-manager

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

+106 −1

Changes for src/srm/adapters/database/repos/runtime_inventory.py: 106 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -14,7 +14,12 @@ from srm.adapters.database.mappers import (
		ServiceInstanceMapper,
		ServiceOrderMapper,
		)
		from srm.adapters.database.sql import CapabilityInstanceRow, ServiceInstanceRow, ServiceOrderRow
		from srm.adapters.database.sql import (
		CapabilityInstanceRow,
		CapabilityRow,
		ServiceInstanceRow,
		ServiceOrderRow,
		)
		from srm.domain.models.runtime_inventory import (
		CapabilityInstance,
		ServiceInstance,
		@@ -25,6 +30,7 @@ from srm.domain.models.runtime_inventory.enums import (
		ServiceInstanceState,
		ServiceOrderState,
		)
		from srm.domain.models.topology.enums import CapabilityKind
		from srm.domain.ports.database.runtime_inventory import (
		CapabilityInstanceRepository,
		ServiceInstanceRepository,
		@@ -96,6 +102,21 @@ class SqlServiceOrderRepository(ServiceOrderRepository):
		raise RuntimeError("Updated service order could not be reloaded")
		return saved

		async def update_federated_handoff(
		self,
		id: UUID,
		federated_handoff: dict[str, object],
		) -> ServiceOrder:
		row = await self._session.get(ServiceOrderRow, id)
		if row is None:
		raise RuntimeError("Service order could not be reloaded")
		row.federated_handoff = federated_handoff
		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:
		@@ -112,6 +133,29 @@ 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_app_provider_id(
		self,
		app_provider_id: str,
		service_instance_id: UUID \| None = None,
		service_specification_id: UUID \| None = None,
		) -> list[ServiceInstance]:
		stmt = (
		select(ServiceInstanceRow)
		.options(
		selectinload(ServiceInstanceRow.capability_instances),
		)
		.where(ServiceInstanceRow.app_provider_id == app_provider_id)
		)
		if service_instance_id is not None:
		stmt = stmt.where(ServiceInstanceRow.id == service_instance_id)
		if service_specification_id is not None:
		stmt = stmt.where(
		ServiceInstanceRow.service_specification_id == service_specification_id
		)

		rows = (await self._session.scalars(stmt)).all()
		return [ServiceInstanceMapper.to_domain(row) for row in rows]

		async def list_by_originating_service_order_id(
		self,
		originating_service_order_id: UUID,
		@@ -201,3 +245,64 @@ class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):
		if saved is None:
		raise RuntimeError("Updated capability instance could not be reloaded")
		return saved

		async def get_by_service_instance_id(
		self,
		service_instance_id: UUID \| None = None,
		) -> list[tuple[CapabilityInstance, CapabilityKind, UUID]]:
		logger.info(
		"get_by_service_instance_id",
		service_instance_id=service_instance_id,
		)
		stmt = (
		select(CapabilityInstanceRow, CapabilityRow.kind, ServiceInstanceRow.zone_id)
		.join(CapabilityRow)
		.where(CapabilityInstanceRow.capability_id == CapabilityRow.id)
		.join(ServiceInstanceRow)
		.where(ServiceInstanceRow.id == service_instance_id)
		.where(CapabilityInstanceRow.service_instance_id == service_instance_id)
		.where(CapabilityInstanceRow.state != CapabilityInstanceState.FAILED)
		.where(CapabilityInstanceRow.state != CapabilityInstanceState.TERMINATED)
		)

		rows = (await self._session.execute(stmt)).all()
		return [(CapabilityInstanceMapper.to_domain(row[0]), row[1], row[2]) for row in rows]

		async def list_by_app_provider_id(
		self,
		app_provider_id: str,
		capability_kind: CapabilityKind \| None = None,
		service_instance_id: UUID \| None = None,
		service_specification_id: UUID \| None = None,
		external_ref: str \| None = None,
		) -> list[tuple[CapabilityInstance, CapabilityKind, UUID]]:
		logger.info(
		"list_by_app_provider_id",
		app_provider_id=app_provider_id,
		capability_kind=capability_kind,
		service_instance_id=service_instance_id,
		service_specification_id=service_specification_id,
		external_ref=external_ref,
		)
		stmt = (
		select(CapabilityInstanceRow, CapabilityRow.kind, ServiceInstanceRow.zone_id)
		.join(CapabilityRow)
		.where(CapabilityInstanceRow.capability_id == CapabilityRow.id)
		.join(ServiceInstanceRow)
		.where(CapabilityInstanceRow.service_instance_id == ServiceInstanceRow.id)
		.where(CapabilityInstanceRow.app_provider_id == app_provider_id)
		.where(CapabilityInstanceRow.state != CapabilityInstanceState.FAILED)
		.where(CapabilityInstanceRow.state != CapabilityInstanceState.TERMINATED)
		)
		if capability_kind is not None:
		stmt = stmt.where(CapabilityRow.kind == capability_kind)
		if service_instance_id is not None:
		stmt = stmt.where(CapabilityInstanceRow.service_instance_id == service_instance_id)
		if service_specification_id is not None:
		stmt = stmt.where(
		ServiceInstanceRow.service_specification_id == service_specification_id
		)
		if external_ref is not None:
		stmt = stmt.where(CapabilityInstanceRow.external_ref == external_ref)
		rows = (await self._session.execute(stmt)).all()
		return [(CapabilityInstanceMapper.to_domain(row[0]), row[1], row[2]) for row in rows]

src/srm/adapters/database/mappers.py

+2 −0

Changes for src/srm/adapters/database/mappers.py: 2 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -318,6 +318,7 @@ class ServiceOrderMapper:
		target_service_instance_id=row.target_service_instance_id,
		state=row.state,
		payload_snapshot=row.payload_snapshot,
		federated_handoff=row.federated_handoff,
		failure_detail=row.failure_detail,
		app_provider_id=row.app_provider_id,
		federation_partner_ref=row.federation_partner_ref,
		@@ -340,6 +341,7 @@ class ServiceOrderMapper:
		target_service_instance_id=domain.target_service_instance_id,
		state=domain.state,
		payload_snapshot=domain.payload_snapshot,
		federated_handoff=domain.federated_handoff,
		failure_detail=domain.failure_detail,
		app_provider_id=domain.app_provider_id,
		federation_partner_ref=domain.federation_partner_ref,

src/srm/adapters/database/sql.py

+3 −0

Changes for src/srm/adapters/database/sql.py: 3 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -270,6 +270,9 @@ class ServiceOrderRow(AuditedMixin, Base):
		)
		state: Mapped[ServiceOrderState] = mapped_column(_enum_type(ServiceOrderState), nullable=False)
		payload_snapshot: Mapped[dict[str, object]] = mapped_column(JSONB, nullable=False, default=dict)
		federated_handoff: Mapped[dict[str, object]] = mapped_column(
		JSONB, nullable=False, default=dict
		)
		failure_detail: Mapped[dict[str, object] \| None] = mapped_column(JSONB)
		app_provider_id: Mapped[str] = mapped_column(String(255), nullable=False)
		federation_partner_ref: Mapped[str \| None] = mapped_column(String(255))

src/srm/adapters/federation_manager/fm_client.py

0 → 100644

+169 −0

Changes for src/srm/adapters/federation_manager/fm_client.py: 169 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""HTTP/Nats Federation Manager adapter."""

		from typing import Any
		from uuid import UUID

		import httpx
		import structlog
		from pydantic import BaseModel

		from srm.adapters.errors import DownstreamServiceException, NotFoundException
		from srm.config import FederationManagerSettings
		from srm.domain.models.runtime_inventory.models import ServiceInstance
		from srm.domain.models.topology.enums import ZoneKind, ZoneState
		from srm.domain.models.topology.models import Zone
		from srm.domain.ports.databus.publisher import DataBusPublisher
		from srm.domain.ports.federation_manager.ports import FederationManagerPort

		_GET_ZONE_ENDPOINT = "/internal/federated-zones/{zone_id}"
		_LIST_ZONES_ENDPOINT = "/internal/federated-zones"
		_GET_SERVICE_INSTANCE_ENDPOINT = "/internal/federated-service-instances/{service_instance_id}"
		_LIST_SERVICE_INSTANCES_ENDPOINT = "/internal/federated-service-instances"


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


		class FederatedZone(BaseModel):
		zone_id: UUID
		partner_op_id: UUID
		supported_apis: list[str] = []
		geography_details: str \| None = None
		geolocation: str \| None = None

		def to_zone(self) -> Zone:
		metadata: dict[str, Any] = {
		"provider": "federation",
		"partner_op_id": str(self.partner_op_id),
		"supported_apis": self.supported_apis,
		}
		if self.geolocation is not None:
		metadata["geolocation"] = self.geolocation
		if self.geography_details is not None:
		metadata["location"] = {"region": self.geography_details}
		return Zone(
		id=self.zone_id,
		ref=f"federated:{self.zone_id}",
		name=self.geography_details or str(self.zone_id),
		kind=ZoneKind.FEDERATION_BUSINESS,
		state=ZoneState.ACTIVE,
		metadata=metadata,
		)


		class FederationManager(FederationManagerPort):
		def __init__(self, settings: FederationManagerSettings, publisher: DataBusPublisher):
		self._settings = settings
		self._publisher = publisher
		logger.info("fm_client_init", settings=self._settings)

		async def _request(
		self,
		method: str,
		path: str,
		json: dict[str, Any] \| None = None,
		params: dict[str, Any] \| None = None,
		headers: dict[str, str] \| None = None,
		) -> Any:
		url = f"{self._settings.base_url}{path}"
		log = logger.bind(
		method=method,
		url=url,
		)

		try:
		async with httpx.AsyncClient(timeout=self._settings.timeout) as client:
		response = await client.request(
		method=method,
		url=url,
		json=json,
		params=params,
		headers=headers,
		)

		if response.status_code == 404:
		log.warning("FM resource not found", url=url)
		raise NotFoundException()

		if response.status_code >= 400:
		log.error("FM returned error", status=response.status_code, body=response.text)
		raise DownstreamServiceException(
		message="FM request failed",
		details={
		"status_code": response.status_code,
		"response": response.text,
		},
		)

		if response.status_code == 204:
		return None

		if not response.content:
		log.error(
		"FM returned empty body for non-204 response", status=response.status_code
		)
		raise DownstreamServiceException(
		message="FM returned empty response",
		details={"status_code": response.status_code},
		)

		return response.json()

		except httpx.TimeoutException as exc:
		log.exception("FM request timed out", error=str(exc))
		raise DownstreamServiceException(
		message="FM request timed out",
		details=str(exc),
		)
		except httpx.ConnectError as exc:
		log.exception("FM connection failed", error=str(exc))
		raise DownstreamServiceException(
		message="Could not connect to FM",
		details=str(exc),
		)
		except httpx.RequestError as exc:
		log.exception("FM request error", error=str(exc))
		raise DownstreamServiceException(
		message="FM request failed",
		details=str(exc),
		)

		async def has_zone(self, zone_id: UUID) -> bool:
		return await self.get_zone(zone_id=zone_id) is not None

		async def get_zone(self, zone_id: UUID) -> Zone \| None:
		try:
		res = await self._request("GET", str.format(_GET_ZONE_ENDPOINT, zone_id=zone_id))
		return FederatedZone.model_validate(res).to_zone()
		except NotFoundException:
		return None

		async def list_zones(self) -> list[Zone]:
		res = await self._request("GET", _LIST_ZONES_ENDPOINT)
		return [FederatedZone.model_validate(obj).to_zone() for obj in res]

		async def list_full_service_instances(self, app_provider_id: str) -> list[ServiceInstance]:
		res = await self._request(
		"GET",
		_LIST_SERVICE_INSTANCES_ENDPOINT,
		params={"app_provider_id": app_provider_id},
		)
		instances = []
		for obj in res:
		instances.append(ServiceInstance.model_validate(obj))
		return instances

		async def get_service_instance(self, service_instance_id: UUID) -> ServiceInstance:
		res = await self._request(
		"GET",
		str.format(_GET_SERVICE_INSTANCE_ENDPOINT, service_instance_id=service_instance_id),
		)

		return ServiceInstance.model_validate(res)

		async def request_deploy_service(self, command: dict[str, Any]) -> None:
		logger.info(
		"deploy_service_on_partner",
		command=command,
		)
		await self._publisher.publish("command.fm.service.deploy", command)

src/srm/adapters/errors.py

+12 −1

Changes for src/srm/adapters/errors.py: 12 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		from typing import ClassVar
		from typing import Any, ClassVar


		class DuplicateEntryError(Exception):
		@@ -15,3 +15,14 @@ class DuplicateServiceSpecificationError(DuplicateEntryError):
		class ServiceSpecificationInUseError(Exception):
		def __init__(self) -> None:
		super().__init__("Service specification has live service instances.")


		class NotFoundException(Exception):
		def __init__(self) -> None:
		super().__init__("Resource Not Found")


		class DownstreamServiceException(Exception):
		def __init__(self, message: str = "Service Unavailable", details: Any \| None = None):
		super().__init__(message)
		self._details = details

SRM Refactor: Federation Manager Client ports and adapters

Merge request reports