feat(fm): outbound federation hand-off (routing, agreements, transactions) (9cc42bd8) · Commits · OOP / code / federation-manager

src/federation_manager/adapters/database/agreement_repo.py

0 → 100644

+42 −0

Original line number	Diff line number	Diff line
		from uuid import UUID

		from sqlalchemy import and_, func, or_, select
		from sqlalchemy.ext.asyncio import AsyncSession

		from federation_manager.adapters.database.tables import federation_agreements as agreements
		from federation_manager.domain.models import Agreement


		class PostgresAgreementRepo:
		def __init__(self, session: AsyncSession) -> None:
		self._session = session

		async def find_active_for_partner(self, partner_id: UUID) -> Agreement \| None:
		now = func.now()
		currently_valid = and_(
		agreements.c.valid_from <= now,
		or_(agreements.c.valid_until.is_(None), agreements.c.valid_until > now),
		)
		# Prefer the agreement valid right now; otherwise the newest, so the domain check
		# reports "expired" against the most relevant contract.
		stmt = (
		select(agreements)
		.where(agreements.c.partner_op_id == partner_id, agreements.c.status == "active")
		.order_by(currently_valid.desc(), agreements.c.valid_from.desc())
		.limit(1)
		)
		row = (await self._session.execute(stmt)).one_or_none()
		if row is None:
		return None
		return Agreement(
		id=row.id,
		partner_op_id=row.partner_op_id,
		permitted_api_types={str(v) for v in row.permitted_api_types},
		permitted_zone_ids={UUID(str(v)) for v in row.permitted_zone_ids},
		service_spec_mappings={
		str(k): UUID(str(v)) for k, v in row.service_spec_mappings.items()
		},
		valid_from=row.valid_from,
		valid_until=row.valid_until,
		status=row.status,
		)

src/federation_manager/adapters/database/partner_repo.py

+11 −2

Original line number	Diff line number	Diff line
		from sqlalchemy import select
		from uuid import UUID

		from sqlalchemy import Select, select
		from sqlalchemy.ext.asyncio import AsyncSession

		from federation_manager.adapters.database.tables import partner_ops
		@@ -10,7 +12,14 @@ class PostgresPartnerRepo:
		self._session = session

		async def find_by_oauth2_client_id(self, client_id: str) -> PartnerOP \| None:
		stmt = select(partner_ops).where(partner_ops.c.oauth2_client_id == client_id)
		return await self._one(
		select(partner_ops).where(partner_ops.c.oauth2_client_id == client_id)
		)

		async def find_by_id(self, partner_id: UUID) -> PartnerOP \| None:
		return await self._one(select(partner_ops).where(partner_ops.c.id == partner_id))

		async def _one(self, stmt: Select[tuple[object, ...]]) -> PartnerOP \| None:
		row = (await self._session.execute(stmt)).one_or_none()
		if row is None:
		return None

src/federation_manager/adapters/database/routing_repo.py

0 → 100644

+32 −0

Original line number	Diff line number	Diff line
		from sqlalchemy import select
		from sqlalchemy.ext.asyncio import AsyncSession

		from federation_manager.adapters.database.tables import routing_rules
		from federation_manager.domain.models import RoutingRule


		class PostgresRoutingRuleRepo:
		def __init__(self, session: AsyncSession) -> None:
		self._session = session

		async def list_active(self, identifier_type: str) -> list[RoutingRule]:
		stmt = (
		select(routing_rules)
		.where(
		routing_rules.c.identifier_type == identifier_type,
		routing_rules.c.is_active.is_(True),
		)
		.order_by(routing_rules.c.priority, routing_rules.c.value_range)
		)
		rows = (await self._session.execute(stmt)).all()
		return [
		RoutingRule(
		id=row.id,
		partner_op_id=row.partner_op_id,
		identifier_type=row.identifier_type,
		value_range=row.value_range,
		priority=row.priority,
		is_active=row.is_active,
		)
		for row in rows
		]

src/federation_manager/adapters/database/tables.py

+86 −1

Original line number	Diff line number	Diff line
		from sqlalchemy import Column, DateTime, Index, MetaData, String, Table, Text, func
		from sqlalchemy import (
		Boolean,
		Column,
		DateTime,
		ForeignKey,
		Index,
		Integer,
		MetaData,
		String,
		Table,
		Text,
		UniqueConstraint,
		func,
		text,
		)
		from sqlalchemy.dialects.postgresql import JSONB
		from sqlalchemy.dialects.postgresql import UUID as PGUUID

		metadata = MetaData()
		@@ -20,3 +35,73 @@ partner_ops = Table(
		Index("idx_partner_ops_oauth2_client_id", "oauth2_client_id"),
		Index("idx_partner_ops_status", "status"),
		)

		# fm_db.federation_agreements (RD §K.1 / PM §J.1).
		federation_agreements = Table(
		"federation_agreements",
		metadata,
		Column("id", PGUUID(as_uuid=True), primary_key=True, server_default=func.gen_random_uuid()),
		Column("partner_op_id", PGUUID(as_uuid=True), ForeignKey("partner_ops.id"), nullable=False),
		Column("permitted_api_types", JSONB, nullable=False),
		Column("permitted_zone_ids", JSONB, nullable=False, server_default=text("'[]'::jsonb")),
		Column("service_spec_mappings", JSONB, nullable=False, server_default=text("'{}'::jsonb")),
		Column("usage_limits", JSONB, server_default=text("'{}'::jsonb")),
		Column("valid_from", DateTime(timezone=True), nullable=False),
		Column("valid_until", DateTime(timezone=True)),
		Column("status", String(20), nullable=False, server_default="draft"),
		Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
		Column("updated_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
		Index("idx_federation_agreements_partner", "partner_op_id"),
		Index("idx_federation_agreements_status_validity", "status", "valid_from", "valid_until"),
		)

		# fm_db.routing_rules (RD §K.1 / PM §J.1).
		routing_rules = Table(
		"routing_rules",
		metadata,
		Column("id", PGUUID(as_uuid=True), primary_key=True, server_default=func.gen_random_uuid()),
		Column("partner_op_id", PGUUID(as_uuid=True), ForeignKey("partner_ops.id"), nullable=False),
		Column("identifier_type", String(20), nullable=False), # msisdn_prefix \| ip_cidr
		Column("value_range", String(50), nullable=False),
		Column("priority", Integer, nullable=False, server_default="100"),
		Column("is_active", Boolean, nullable=False, server_default=text("true")),
		Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
		Column("updated_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
		UniqueConstraint("identifier_type", "value_range", "priority", name="uq_routing_rules_rule"),
		Index("idx_routing_rules_type_value", "identifier_type", "value_range"),
		Index("idx_routing_rules_partner", "partner_op_id"),
		Index("idx_routing_rules_active", "is_active", postgresql_where=text("is_active = true")),
		)

		# fm_db.federation_transactions (RD §K.1 / PM §J.1).
		federation_transactions = Table(
		"federation_transactions",
		metadata,
		Column("id", PGUUID(as_uuid=True), primary_key=True, server_default=func.gen_random_uuid()),
		Column("partner_op_id", PGUUID(as_uuid=True), ForeignKey("partner_ops.id"), nullable=False),
		Column("agreement_id", PGUUID(as_uuid=True), ForeignKey("federation_agreements.id")),
		Column("direction", String(10), nullable=False), # inbound \| outbound
		Column("federation_operation_id", PGUUID(as_uuid=True)),
		Column("operation_id", PGUUID(as_uuid=True)),
		Column("correlation_id", PGUUID(as_uuid=True)),
		Column("federation_correlation_id", String(255), nullable=False),
		Column("api_type", String(100), nullable=False),
		Column("status", String(20), nullable=False, server_default="pending"),
		Column("request_summary", JSONB, nullable=False),
		Column("response_summary", JSONB),
		Column("error_detail", JSONB),
		Column("started_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
		Column("completed_at", DateTime(timezone=True)),
		Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
		Index("idx_fed_tx_partner", "partner_op_id"),
		Index(
		"uq_fed_tx_federation_operation_id",
		"federation_operation_id",
		unique=True,
		postgresql_where=text("federation_operation_id IS NOT NULL"),
		),
		Index("idx_fed_tx_operation", "operation_id"),
		Index("idx_fed_tx_status", "status"),
		Index("idx_fed_tx_started", "started_at"),
		Index("idx_fed_tx_fed_correlation", "federation_correlation_id"),
		)

src/federation_manager/adapters/database/transaction_repo.py

0 → 100644

+56 −0

Original line number	Diff line number	Diff line
		from datetime import datetime
		from uuid import UUID

		from sqlalchemy import insert, update
		from sqlalchemy.ext.asyncio import AsyncSession

		from federation_manager.adapters.database.tables import federation_transactions as transactions
		from federation_manager.domain.models import FederationTransaction


		class PostgresTransactionRepo:
		"""Audit rows commit immediately: they must survive whatever happens on the partner call."""

		def __init__(self, session: AsyncSession) -> None:
		self._session = session

		async def add(self, transaction: FederationTransaction) -> None:
		await self._session.execute(
		insert(transactions).values(
		id=transaction.id,
		partner_op_id=transaction.partner_op_id,
		agreement_id=transaction.agreement_id,
		direction=transaction.direction,
		federation_operation_id=transaction.federation_operation_id,
		operation_id=transaction.operation_id,
		correlation_id=transaction.correlation_id,
		federation_correlation_id=transaction.federation_correlation_id,
		api_type=transaction.api_type,
		status=transaction.status,
		request_summary=transaction.request_summary,
		response_summary=transaction.response_summary,
		error_detail=transaction.error_detail,
		started_at=transaction.started_at,
		completed_at=transaction.completed_at,
		)
		)
		await self._session.commit()

		async def record_outcome(
		self,
		transaction_id: UUID,
		*,
		status: str,
		completed_at: datetime,
		response_summary: dict[str, object] \| None = None,
		error_detail: dict[str, object] \| None = None,
		) -> None:
		values: dict[str, object] = {"status": status, "completed_at": completed_at}
		if response_summary is not None:
		values["response_summary"] = response_summary
		if error_detail is not None:
		values["error_detail"] = error_detail
		await self._session.execute(
		update(transactions).where(transactions.c.id == transaction_id).values(**values)
		)
		await self._session.commit()