feat(fm): store the zones a partner offers (#23) (!23) · Merge requests · OOP / code / federation-manager

src/federation_manager/adapters/database/catalogue_repo.py

0 → 100644

+85 −0

Changes for src/federation_manager/adapters/database/catalogue_repo.py: 85 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from typing import Any
		from uuid import UUID

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

		from federation_manager.adapters.database.tables import shared_resource_catalogues as catalogues
		from federation_manager.domain.models import INBOUND, FederatedZone
		from federation_manager.domain.zones import EDGE_ZONE


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

		async def replace_partner_zones(self, partner_id: UUID, zones: list[FederatedZone]) -> None:
		"""Make the stored catalogue match the zone list the partner just sent.

		The partner always offers its whole catalogue, so the rows are replaced rather than
		merged: zones it dropped disappear, and because zone_id is derived the ids of the
		zones it kept do not change.
		"""
		await self._session.execute(
		delete(catalogues).where(
		catalogues.c.partner_op_id == partner_id,
		catalogues.c.resource_type == EDGE_ZONE,
		)
		)
		if zones:
		await self._session.execute(
		insert(catalogues),
		[
		{
		"partner_op_id": partner_id,
		"direction": INBOUND,
		"resource_type": EDGE_ZONE,
		"resource_data": _data(zone),
		}
		for zone in zones
		],
		)
		await self._session.commit()

		async def find_zone(self, zone_id: UUID) -> FederatedZone \| None:
		return await self._one(
		select(catalogues).where(
		catalogues.c.resource_type == EDGE_ZONE,
		catalogues.c.resource_data["zone_id"].astext == str(zone_id),
		)
		)

		async def list_zones(self) -> list[FederatedZone]:
		stmt = (
		select(catalogues)
		.where(catalogues.c.resource_type == EDGE_ZONE)
		.order_by(catalogues.c.created_at)
		)
		rows = (await self._session.execute(stmt)).all()
		return [_to_zone(row) for row in rows]

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


		def _data(zone: FederatedZone) -> dict[str, Any]:
		return {
		"zone_id": str(zone.zone_id),
		"external_zone_id": zone.external_zone_id,
		"geography_details": zone.geography_details,
		"geolocation": zone.geolocation,
		"supported_apis": list(zone.supported_apis),
		}


		def _to_zone(row: Any) -> FederatedZone:
		data = row.resource_data
		return FederatedZone(
		zone_id=UUID(data["zone_id"]),
		partner_op_id=row.partner_op_id,
		external_zone_id=data["external_zone_id"],
		geography_details=data.get("geography_details"),
		geolocation=data.get("geolocation"),
		supported_apis=tuple(data.get("supported_apis") or ()),
		)

src/federation_manager/adapters/database/tables.py

+31 −0

Changes for src/federation_manager/adapters/database/tables.py: 31 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -153,6 +153,37 @@ routing_rules = Table(
		),
		)

		# direction follows the resource, not the federation: a zone a partner offers us is `inbound`.
		shared_resource_catalogues = Table(
		"shared_resource_catalogues",
		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("direction", String(10), nullable=False),
		Column("resource_type", String(50), nullable=False),
		Column("resource_data", JSONB, nullable=False),
		Column("synced_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
		Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
		Column(
		"updated_at",
		DateTime(timezone=True),
		nullable=False,
		server_default=func.now(),
		onupdate=func.now(),
		),
		Index("idx_shared_catalogue_partner_direction", "partner_op_id", "direction"),
		Index("idx_shared_catalogue_resource_type", "resource_type"),
		# one row per federated zone: zone_id is derived, so re-syncing is an upsert on this key
		Index(
		"uq_shared_catalogue_zone",
		text("(resource_data ->> 'zone_id')"),
		unique=True,
		postgresql_where=text("resource_type = 'edge_zone'"),
		),
		CheckConstraint(_one_of("direction", *DIRECTIONS), name="ck_shared_catalogue_direction"),
		)


		federation_transactions = Table(
		"federation_transactions",
		metadata,

src/federation_manager/adapters/http/srm_client.py

+25 −0

Changes for src/federation_manager/adapters/http/srm_client.py: 25 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from uuid import UUID

		import httpx

		from federation_manager.contracts.srm import LocationQueryRequestV1, LocationQueryResponseV1
		from federation_manager.domain.errors import NetworkQueryNotApplicable, SrmQueryFailed
		from federation_manager.domain.models import LocalZone

		LOCATION_PATH = "/internal/network-queries/location"
		ZONES_PATH = "/internal/zones"


		class HttpxSrmClient:
		@@ -23,3 +27,24 @@ class HttpxSrmClient:
		return LocationQueryResponseV1.model_validate(response.json())
		except (httpx.HTTPError, ValueError) as error:
		raise SrmQueryFailed(str(error)[:200]) from None

		async def list_zones(self) -> list[LocalZone]:
		try:
		response = await self._client.get(
		f"{self._base_url}{ZONES_PATH}", params={"state": "active"}
		)
		response.raise_for_status()
		return [_local_zone(zone) for zone in response.json()]
		except (httpx.HTTPError, ValueError, KeyError, TypeError) as error:
		raise SrmQueryFailed(str(error)[:200]) from None


		def _local_zone(zone: dict[str, object]) -> LocalZone:
		metadata = zone.get("metadata")
		location = metadata.get("location") if isinstance(metadata, dict) else None
		region = location.get("region") if isinstance(location, dict) else None
		return LocalZone(
		id=UUID(str(zone["id"])),
		name=str(zone["name"]),
		region=str(region) if region is not None else None,
		)

src/federation_manager/api/internal/zones.py

0 → 100644

+46 −0

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

		from fastapi import APIRouter, Depends
		from pydantic import BaseModel

		from federation_manager.dependencies import get_catalogue_repo
		from federation_manager.domain.errors import ZoneNotFederated
		from federation_manager.domain.models import FederatedZone
		from federation_manager.domain.ports import CatalogueRepositoryPort

		# SRM reads these to tell a partner's zone from an unknown one, and to list partner zones.
		router = APIRouter(prefix="/internal/federated-zones", tags=["internal-federated-zones"])

		Catalogue = Annotated[CatalogueRepositoryPort, Depends(get_catalogue_repo)]


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

		@classmethod
		def of(cls, zone: FederatedZone) -> "FederatedZoneView":
		return cls(
		zone_id=zone.zone_id,
		partner_op_id=zone.partner_op_id,
		supported_apis=list(zone.supported_apis),
		geography_details=zone.geography_details,
		geolocation=zone.geolocation,
		)


		@router.get("", response_model=list[FederatedZoneView])
		async def list_federated_zones(catalogue: Catalogue) -> list[FederatedZoneView]:
		return [FederatedZoneView.of(zone) for zone in await catalogue.list_zones()]


		@router.get("/{zone_id}", response_model=FederatedZoneView)
		async def get_federated_zone(zone_id: UUID, catalogue: Catalogue) -> FederatedZoneView:
		zone = await catalogue.find_zone(zone_id)
		if zone is None:
		raise ZoneNotFederated(zone_id)
		return FederatedZoneView.of(zone)

src/federation_manager/api/errors.py

+11 −0

Changes for src/federation_manager/api/errors.py: 11 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -31,6 +31,7 @@ from federation_manager.domain.errors import (
		SessionUnknown,
		SrmQueryFailed,
		UnsupportedServiceApi,
		ZoneNotFederated,
		problem_type,
		)

		@@ -347,3 +348,13 @@ def register_exception_handlers(app: FastAPI) -> None:
		str(exc),
		request.url.path,
		)

		@app.exception_handler(ZoneNotFederated)
		async def _zone_not_federated(request: Request, exc: ZoneNotFederated) -> JSONResponse:
		return problem(
		404,
		"zone-not-federated",
		"Zone Not Federated",
		"No federation partner offers this zone.",
		request.url.path,
		)

feat(fm): store the zones a partner offers (#23)

Merge request reports