Feat/app onboarding (!25) · Merge requests · OOP / code / federation-manager

src/federation_manager/adapters/database/catalogue_repo.py

+55 −0

Changes for src/federation_manager/adapters/database/catalogue_repo.py: 55 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -109,6 +109,53 @@ class PostgresCatalogueRepo:
		await self._session.commit()
		return found

		async def put_resource(
		self, partner_id: UUID, direction: str, resource_type: str, key: str, data: dict[str, Any]
		) -> None:
		await self._session.execute(
		delete(catalogues).where(_keyed(partner_id, resource_type, key))
		)
		await self._session.execute(
		insert(catalogues).values(
		partner_op_id=partner_id,
		direction=direction,
		resource_type=resource_type,
		resource_data={**data, "key": key},
		)
		)
		await self._session.commit()

		async def get_resource(
		self, partner_id: UUID, resource_type: str, key: str
		) -> dict[str, Any] \| None:
		row = (
		await self._session.execute(
		select(catalogues).where(_keyed(partner_id, resource_type, key))
		)
		).one_or_none()
		return None if row is None else dict(row.resource_data)

		async def list_resources(self, partner_id: UUID, resource_type: str) -> list[dict[str, Any]]:
		stmt = (
		select(catalogues)
		.where(
		catalogues.c.partner_op_id == partner_id,
		catalogues.c.resource_type == resource_type,
		)
		.order_by(catalogues.c.created_at)
		)
		return [dict(row.resource_data) for row in (await self._session.execute(stmt)).all()]

		async def delete_resource(self, partner_id: UUID, resource_type: str, key: str) -> bool:
		removed = await self._session.execute(
		delete(catalogues)
		.where(_keyed(partner_id, resource_type, key))
		.returning(catalogues.c.id)
		)
		found = removed.first() is not None
		await self._session.commit()
		return found

		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)
		@@ -150,3 +197,11 @@ def _to_subscription(row: Any) -> ZoneSubscription:
		zone_id=UUID(row.resource_data["zone_id"]),
		avail_zone_notif_link=row.resource_data.get("avail_zone_notif_link"),
		)


		def _keyed(partner_id: UUID, resource_type: str, key: str) -> Any:
		return and_(
		catalogues.c.partner_op_id == partner_id,
		catalogues.c.resource_type == resource_type,
		catalogues.c.resource_data["key"].astext == key,
		)

src/federation_manager/adapters/database/tables.py

+9 −0

Changes for src/federation_manager/adapters/database/tables.py: 9 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -190,6 +190,15 @@ shared_resource_catalogues = Table(
		unique=True,
		postgresql_where=text("resource_type = 'zone_subscription'"),
		),
		# every other exchanged resource: one row per partner, type and key
		Index(
		"uq_shared_catalogue_keyed_resource",
		"partner_op_id",
		"resource_type",
		text("(resource_data ->> 'key')"),
		unique=True,
		postgresql_where=text("resource_data ? 'key'"),
		),
		CheckConstraint(_one_of("direction", *DIRECTIONS), name="ck_shared_catalogue_direction"),
		)

src/federation_manager/adapters/databus/nats_adapter.py

+29 −12

Changes for src/federation_manager/adapters/databus/nats_adapter.py: 29 added lines, 12 removed lines.
Original line number	Diff line number	Diff line
		@@ -41,7 +41,7 @@ class NatsCommandPublisher:
		js = self._require_js()
		config = StreamConfig(
		name=TASK_STREAM,
		subjects=["command.srm.>"],
		subjects=["command.srm.>", "command.fm.>"],
		retention=RetentionPolicy.WORK_QUEUE,
		max_age=TASK_STREAM_MAX_AGE_SECONDS,
		)
		@@ -82,16 +82,18 @@ class NatsEventConsumer:

		async def ensure_event_stream(self) -> None:
		js = self._require_js()
		try:
		await js.stream_info(EVENT_STREAM)
		except Exception:
		await js.add_stream(
		StreamConfig(
		config = StreamConfig(
		name=EVENT_STREAM,
		subjects=["event.srm.>"],
		subjects=["event.srm.>", "event.fm.>"],
		retention=RetentionPolicy.LIMITS,
		)
		)
		try:
		await js.stream_info(EVENT_STREAM)
		except Exception:
		await js.add_stream(config)
		else:
		# a stream from an older FM lacks event.fm.>, and FM's own publishes would fail
		await js.update_stream(config)

		async def subscribe(
		self, subject: str, handler: Callable[[dict[str, Any]], Awaitable[None]]
		@@ -110,23 +112,38 @@ class NatsEventConsumer:
		return
		await message.ack()

		name = f"{self._durable}-{subject.replace('.', '-')}"
		name = self._consumer_name(subject)
		# queue group so replicas of this FM share one durable instead of fighting over it
		await self._require_js().subscribe(
		subject,
		queue=name,
		durable=name,
		stream=EVENT_STREAM,
		stream=_stream_for(subject),
		cb=on_message,
		manual_ack=True,
		config=ConsumerConfig(max_deliver=3, ack_policy=AckPolicy.EXPLICIT),
		)

		def _consumer_name(self, subject: str) -> str:
		# A work-queue stream allows one consumer per subject, so commands share one durable
		# (ADR-0053's fm-task-worker); events keep a per-FM durable each.
		if subject.startswith("command."):
		return TASK_CONSUMER
		return f"{self._durable}-{subject.replace('.', '-')}"

		async def delete_durable(self, subject: str) -> None:
		name = f"{self._durable}-{subject.replace('.', '-')}"
		await self._require_js().delete_consumer(EVENT_STREAM, name)
		name = self._consumer_name(subject)
		await self._require_js().delete_consumer(_stream_for(subject), name)

		def _require_js(self) -> JetStreamContext:
		if self._js is None:
		raise RuntimeError("NATS event consumer is not connected")
		return self._js


		TASK_CONSUMER = "fm-task-worker"


		def _stream_for(subject: str) -> str:
		# commands live in the work-queue stream, events in the event stream
		return TASK_STREAM if subject.startswith("command.") else EVENT_STREAM

src/federation_manager/adapters/http/ewbi_client.py

+31 −3

Changes for src/federation_manager/adapters/http/ewbi_client.py: 31 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		import json

		import httpx

		from federation_manager.domain.errors import (
		@@ -19,8 +21,27 @@ class HttpxEwbiClient:
		self._token_provider = token_provider
		self._allow_insecure = allow_insecure

		async def post(self, partner: PartnerOP, path: str, payload: dict[str, object]) -> EwbiResponse:
		return await self._send(partner, "POST", path, payload=payload)
		async def post(
		self,
		partner: PartnerOP,
		path: str,
		payload: dict[str, object],
		headers: dict[str, str] \| None = None,
		) -> EwbiResponse:
		return await self._send(partner, "POST", path, payload=payload, headers=headers)

		async def post_form(
		self, partner: PartnerOP, path: str, fields: dict[str, object]
		) -> EwbiResponse:
		# multipart/form-data; object parts go as JSON, OpenAPI's default for them
		parts = {
		name: (None, value, "text/plain")
		if isinstance(value, str)
		else (None, json.dumps(value), "application/json")
		for name, value in fields.items()
		if value is not None
		}
		return await self._send(partner, "POST", path, files=parts)

		async def get(
		self, partner: PartnerOP, path: str, params: dict[str, str] \| None = None
		@@ -38,6 +59,8 @@ class HttpxEwbiClient:
		*,
		payload: dict[str, object] \| None = None,
		params: dict[str, str] \| None = None,
		files: dict[str, tuple[None, str, str]] \| None = None,
		headers: dict[str, str] \| None = None,
		) -> EwbiResponse:
		url = self._url(partner, path)
		token = await self._token_provider.token_for(partner, scope="fed-mgmt")
		@@ -46,9 +69,14 @@ class HttpxEwbiClient:
		response = await self._client.request(
		method,
		url,
		headers={"Accept": "application/json", "Authorization": f"Bearer {token}"},
		headers={
		**(headers or {}),
		"Accept": "application/json",
		"Authorization": f"Bearer {token}",
		},
		json=payload,
		params=params,
		files=files,
		)
		except httpx.HTTPError:
		raise PartnerRequestFailed(partner.id) from None

src/federation_manager/adapters/http/srm_client.py

+31 −1

Changes for src/federation_manager/adapters/http/srm_client.py: 31 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		from typing import Any
		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.errors import (
		NetworkQueryNotApplicable,
		ServiceSpecificationUnknown,
		SrmQueryFailed,
		)
		from federation_manager.domain.models import LocalZone

		LOCATION_PATH = "/internal/network-queries/location"
		ZONES_PATH = "/internal/zones"
		SPECIFICATIONS_PATH = "/internal/catalog/service-specifications"


		class HttpxSrmClient:
		@@ -38,6 +44,30 @@ class HttpxSrmClient:
		except (httpx.HTTPError, ValueError, KeyError, TypeError) as error:
		raise SrmQueryFailed(str(error)[:200]) from None

		async def get_service_specification(self, specification_id: UUID) -> dict[str, Any]:
		try:
		response = await self._client.get(
		f"{self._base_url}{SPECIFICATIONS_PATH}/{specification_id}"
		)
		if response.status_code == 404:
		raise ServiceSpecificationUnknown(specification_id)
		response.raise_for_status()
		body: dict[str, Any] = response.json()
		return body
		except (httpx.HTTPError, ValueError) as error:
		raise SrmQueryFailed(str(error)[:200]) from None

		async def create_service_specification(self, payload: dict[str, Any]) -> None:
		try:
		response = await self._client.post(
		f"{self._base_url}{SPECIFICATIONS_PATH}", json=payload
		)
		# 409: an earlier attempt already created this specification id
		if response.status_code != 409:
		response.raise_for_status()
		except httpx.HTTPError as error:
		raise SrmQueryFailed(str(error)[:200]) from None


		def _local_zone(zone: dict[str, object]) -> LocalZone:
		metadata = zone.get("metadata")

Feat/app onboarding

Merge request reports