feat(fm): read partner-hosted service instances for SRM (de23eeb3) · Commits · OOP / code / federation-manager

src/federation_manager/adapters/database/transaction_repo.py

+44 −25

Changes for src/federation_manager/adapters/database/transaction_repo.py: 44 added lines, 25 removed lines.
Original line number	Diff line number	Diff line
		from datetime import datetime
		from typing import Any
		from uuid import UUID

		from sqlalchemy import Select, insert, select, update
		from sqlalchemy import Row, Select, insert, select, update
		from sqlalchemy.ext.asyncio import AsyncSession

		from federation_manager.adapters.database.tables import federation_transactions as transactions
		@@ -88,34 +89,25 @@ class PostgresTransactionRepo:
		.limit(1)
		)

		async def list_outbound_by_app_provider(
		self, api_type: str, app_provider_id: str
		) -> list[FederationTransaction]:
		rows = await self._session.execute(
		select(transactions)
		.where(
		transactions.c.direction == OUTBOUND,
		transactions.c.api_type == api_type,
		transactions.c.request_summary["app_provider_id"].astext == app_provider_id,
		)
		.order_by(transactions.c.started_at)
		)
		return [_transaction(row) for row in rows]

		async def _one(self, stmt: Select[tuple[object, ...]]) -> FederationTransaction \| None:
		row = (await self._session.execute(stmt)).one_or_none()
		if row is None:
		return None
		return FederationTransaction(
		id=row.id,
		partner_op_id=row.partner_op_id,
		direction=row.direction,
		api_type=row.api_type,
		status=row.status,
		request_summary=row.request_summary,
		started_at=row.started_at,
		agreement_id=row.agreement_id,
		federation_context_row_id=row.federation_context_row_id,
		external_txn_id=row.external_txn_id,
		idempotency_key=row.idempotency_key,
		request_fingerprint=row.request_fingerprint,
		external_resource_id=row.external_resource_id,
		callback_url=row.callback_url,
		callback_status=row.callback_status,
		callback_attempts=row.callback_attempts,
		operation_id=row.operation_id,
		correlation_id=row.correlation_id,
		federation_operation_id=row.federation_operation_id,
		response_summary=row.response_summary,
		error_detail=row.error_detail,
		completed_at=row.completed_at,
		)
		return _transaction(row)

		async def mark_in_progress(self, transaction_id: UUID) -> None:
		await self._session.execute(
		@@ -154,3 +146,30 @@ class PostgresTransactionRepo:
		update(transactions).where(transactions.c.id == transaction_id).values(**values)
		)
		await self._session.commit()


		def _transaction(row: Row[Any]) -> FederationTransaction:
		return FederationTransaction(
		id=row.id,
		partner_op_id=row.partner_op_id,
		direction=row.direction,
		api_type=row.api_type,
		status=row.status,
		request_summary=row.request_summary,
		started_at=row.started_at,
		agreement_id=row.agreement_id,
		federation_context_row_id=row.federation_context_row_id,
		external_txn_id=row.external_txn_id,
		idempotency_key=row.idempotency_key,
		request_fingerprint=row.request_fingerprint,
		external_resource_id=row.external_resource_id,
		callback_url=row.callback_url,
		callback_status=row.callback_status,
		callback_attempts=row.callback_attempts,
		operation_id=row.operation_id,
		correlation_id=row.correlation_id,
		federation_operation_id=row.federation_operation_id,
		response_summary=row.response_summary,
		error_detail=row.error_detail,
		completed_at=row.completed_at,
		)

src/federation_manager/api/errors.py

+13 −0

Changes for src/federation_manager/api/errors.py: 13 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -13,6 +13,7 @@ from federation_manager.domain.errors import (
		AgreementViolation,
		AppInstanceUnknown,
		AuthenticationFailed,
		FederatedInstanceUnknown,
		FederationAlreadyExists,
		FederationContextMissing,
		FederationContextUnknown,
		@@ -437,6 +438,18 @@ def register_exception_handlers(app: FastAPI) -> None:
		request.url.path,
		)

		@app.exception_handler(FederatedInstanceUnknown)
		async def _federated_instance_unknown(
		request: Request, exc: FederatedInstanceUnknown
		) -> JSONResponse:
		return problem(
		404,
		"instance-not-federated",
		"Instance Not Federated",
		"No partner hosts a service instance with this identifier.",
		request.url.path,
		)

		@app.exception_handler(InstanceUnknown)
		async def _instance_unknown(request: Request, exc: InstanceUnknown) -> JSONResponse:
		return problem(

src/federation_manager/api/internal/service_instances.py

0 → 100644

+76 −0

Changes for src/federation_manager/api/internal/service_instances.py: 76 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from datetime import datetime
		from typing import Annotated, Any
		from uuid import UUID

		from fastapi import APIRouter, Depends
		from pydantic import BaseModel, ConfigDict, Field

		from federation_manager.application.federated_instances import (
		FederatedInstance,
		FederatedInstanceQueries,
		State,
		)
		from federation_manager.dependencies import get_transaction_repo
		from federation_manager.domain.errors import FederatedInstanceUnknown
		from federation_manager.domain.ports import TransactionRepositoryPort

		# SRM merges these into its instance listings: it keeps no rows for partner-hosted instances.
		router = APIRouter(
		prefix="/internal/federated-service-instances", tags=["internal-federated-service-instances"]
		)

		Transactions = Annotated[TransactionRepositoryPort, Depends(get_transaction_repo)]


		class FederatedServiceInstanceView(BaseModel):
		"""Shaped as SRM's own service instance, which SRM validates it against."""

		model_config = ConfigDict(extra="forbid")

		id: UUID
		service_specification_id: UUID
		zone_id: UUID
		ref: str
		state: State
		app_provider_id: str
		# SRM's field names the partner a deploy came from; these went out from us
		federation_partner_ref: str \| None = None
		metadata: dict[str, Any] = Field(default_factory=dict)
		capabilities: list[dict[str, Any]] = Field(default_factory=list)
		created_at: datetime
		updated_at: datetime
		terminated_at: datetime \| None = None

		@classmethod
		def of(cls, instance: FederatedInstance) -> "FederatedServiceInstanceView":
		return cls(
		id=instance.service_instance_id,
		service_specification_id=instance.service_specification_id,
		zone_id=instance.zone_id,
		# the partner's appInstIdentifier stays with FM, as its zone ids do
		ref=f"federated:{instance.service_instance_id}",
		state=instance.state,
		app_provider_id=instance.app_provider_id,
		metadata={"provider": "federation", "partner_op_id": str(instance.partner_op_id)},
		created_at=instance.created_at,
		updated_at=instance.updated_at,
		terminated_at=instance.terminated_at,
		)


		@router.get("", response_model=list[FederatedServiceInstanceView])
		async def list_federated_service_instances(
		app_provider_id: str, transactions: Transactions
		) -> list[FederatedServiceInstanceView]:
		instances = await FederatedInstanceQueries(transactions).list(app_provider_id)
		return [FederatedServiceInstanceView.of(instance) for instance in instances]


		@router.get("/{service_instance_id}", response_model=FederatedServiceInstanceView)
		async def get_federated_service_instance(
		service_instance_id: UUID, transactions: Transactions
		) -> FederatedServiceInstanceView:
		instance = await FederatedInstanceQueries(transactions).get(service_instance_id)
		if instance is None:
		raise FederatedInstanceUnknown(service_instance_id)
		return FederatedServiceInstanceView.of(instance)

src/federation_manager/application/federated_handoff.py

+6 −1

Changes for src/federation_manager/application/federated_handoff.py: 6 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -103,7 +103,12 @@ class FederatedDeployHandler:
		context.federation_context_id,
		command.service_specification_id,
		zone.zone_id,
		SrmHandOff(command.operation_id, command.correlation_id, service_instance_id),
		SrmHandOff(
		command.operation_id,
		command.correlation_id,
		service_instance_id,
		command.app_provider_id,
		),
		)

src/federation_manager/application/federated_instances.py

0 → 100644

+84 −0

Changes for src/federation_manager/application/federated_instances.py: 84 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from dataclasses import dataclass
		from datetime import datetime
		from typing import Literal
		from uuid import UUID

		from federation_manager.application.federated_handoff import REMOVE_APP
		from federation_manager.application.outbound_deploy import INSTALL_APP
		from federation_manager.domain.models import FederationTransaction
		from federation_manager.domain.ports import TransactionRepositoryPort

		# SRM's ServiceInstanceState values a partner-hosted instance can be in
		State = Literal["creating", "active", "failed", "terminated"]


		@dataclass(frozen=True)
		class FederatedInstance:
		service_instance_id: UUID
		service_specification_id: UUID
		zone_id: UUID
		app_provider_id: str
		partner_op_id: UUID
		state: State
		created_at: datetime
		updated_at: datetime
		terminated_at: datetime \| None


		class FederatedInstanceQueries:
		"""SRM keeps no rows for instances a partner hosts (ADR-0053), so it reads them here.

		An instance is the InstallApp we sent for SRM's hand-off, plus the RemoveApp that ended
		it, if any. Only installs SRM handed us count: they carry its instance id and App
		Provider.
		"""

		def __init__(self, transactions: TransactionRepositoryPort) -> None:
		self._transactions = transactions

		async def get(self, service_instance_id: UUID) -> FederatedInstance \| None:
		install = await self._transactions.find_outbound_by_service_instance(
		INSTALL_APP, service_instance_id
		)
		if install is None or "app_provider_id" not in install.request_summary:
		return None
		return await self._instance(install)

		async def list(self, app_provider_id: str) -> list[FederatedInstance]:
		installs = await self._transactions.list_outbound_by_app_provider(
		INSTALL_APP, app_provider_id
		)
		# a retried hand-off can leave several installs for one instance; the latest wins
		latest = {str(t.request_summary["service_instance_id"]): t for t in installs}
		return [await self._instance(install) for install in latest.values()]

		async def _instance(self, install: FederationTransaction) -> FederatedInstance:
		summary = install.request_summary
		service_instance_id = UUID(str(summary["service_instance_id"]))
		removal = await self._transactions.find_outbound_by_service_instance(
		REMOVE_APP, service_instance_id
		)
		if removal is not None and removal.started_at < install.started_at:
		removal = None # ended an earlier install of the same instance
		state: State
		if removal is not None:
		# OPG.04 gives RemoveApp no callback: the partner accepting it is the end
		state = "terminated"
		elif install.status == "completed":
		state = "active"
		elif install.status == "failed":
		state = "failed"
		else:
		state = "creating"
		terminated_at = (removal.completed_at or removal.started_at) if removal else None
		return FederatedInstance(
		service_instance_id=service_instance_id,
		service_specification_id=UUID(str(summary["service_specification_id"])),
		zone_id=UUID(str(summary["zone_id"])),
		app_provider_id=str(summary["app_provider_id"]),
		partner_op_id=install.partner_op_id,
		state=state,
		created_at=install.started_at,
		updated_at=terminated_at or install.completed_at or install.started_at,
		terminated_at=terminated_at,
		)