mplement zone subscription flow (!24) · Merge requests · OOP / code / federation-manager

src/federation_manager/adapters/database/agreement_repo.py

+1 −0

Changes for src/federation_manager/adapters/database/agreement_repo.py: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -49,4 +49,5 @@ class PostgresAgreementRepo:
		valid_from=row.valid_from,
		valid_until=row.valid_until,
		status=row.status,
		zone_resources={UUID(str(k)): v for k, v in (row.zone_resources or {}).items()},
		)

src/federation_manager/adapters/database/catalogue_repo.py

+70 −3

Changes for src/federation_manager/adapters/database/catalogue_repo.py: 70 added lines, 3 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 import Select, and_, delete, insert, or_, 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
		from federation_manager.domain.models import INBOUND, OUTBOUND, FederatedZone, ZoneSubscription
		from federation_manager.domain.zones import EDGE_ZONE, ZONE_SUBSCRIPTION


		class PostgresCatalogueRepo:
		@@ -58,6 +58,57 @@ class PostgresCatalogueRepo:
		rows = (await self._session.execute(stmt)).all()
		return [_to_zone(row) for row in rows]

		async def subscribe(self, subscriptions: list[ZoneSubscription]) -> None:
		# replace, so a repeated ZoneSubscribe keeps one row and picks up a new notif link
		await self._session.execute(
		delete(catalogues).where(
		or_(*(_subscription(sub.partner_op_id, sub.zone_id) for sub in subscriptions))
		)
		)
		await self._session.execute(
		insert(catalogues),
		[
		{
		"partner_op_id": sub.partner_op_id,
		"direction": OUTBOUND,
		"resource_type": ZONE_SUBSCRIPTION,
		"resource_data": {
		"zone_id": str(sub.zone_id),
		"avail_zone_notif_link": sub.avail_zone_notif_link,
		},
		}
		for sub in subscriptions
		],
		)
		await self._session.commit()

		async def find_subscription(self, partner_id: UUID, zone_id: UUID) -> ZoneSubscription \| None:
		row = (
		await self._session.execute(
		select(catalogues).where(_subscription(partner_id, zone_id))
		)
		).one_or_none()
		return None if row is None else _to_subscription(row)

		async def list_subscriptions(self, partner_id: UUID) -> list[ZoneSubscription]:
		stmt = (
		select(catalogues)
		.where(
		catalogues.c.partner_op_id == partner_id,
		catalogues.c.resource_type == ZONE_SUBSCRIPTION,
		)
		.order_by(catalogues.c.created_at)
		)
		return [_to_subscription(row) for row in (await self._session.execute(stmt)).all()]

		async def unsubscribe(self, partner_id: UUID, zone_id: UUID) -> bool:
		removed = await self._session.execute(
		delete(catalogues).where(_subscription(partner_id, zone_id)).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)
		@@ -83,3 +134,19 @@ def _to_zone(row: Any) -> FederatedZone:
		geolocation=data.get("geolocation"),
		supported_apis=tuple(data.get("supported_apis") or ()),
		)


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


		def _to_subscription(row: Any) -> ZoneSubscription:
		return ZoneSubscription(
		partner_op_id=row.partner_op_id,
		zone_id=UUID(row.resource_data["zone_id"]),
		avail_zone_notif_link=row.resource_data.get("avail_zone_notif_link"),
		)

src/federation_manager/adapters/database/tables.py

+10 −0

Changes for src/federation_manager/adapters/database/tables.py: 10 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -68,6 +68,8 @@ federation_agreements = Table(
		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")),
		# per zone id: what this partner gets there (GSMA reserved resources, quota, flavours)
		Column("zone_resources", JSONB, nullable=False, 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"),
		@@ -180,6 +182,14 @@ shared_resource_catalogues = Table(
		unique=True,
		postgresql_where=text("resource_type = 'edge_zone'"),
		),
		# a partner subscribes to each of our zones at most once
		Index(
		"uq_shared_catalogue_zone_subscription",
		"partner_op_id",
		text("(resource_data ->> 'zone_id')"),
		unique=True,
		postgresql_where=text("resource_type = 'zone_subscription'"),
		),
		CheckConstraint(_one_of("direction", *DIRECTIONS), name="ck_shared_catalogue_direction"),
		)

src/federation_manager/api/ewbi/v1/zones.py

0 → 100644

+93 −0

Changes for src/federation_manager/api/ewbi/v1/zones.py: 93 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from typing import Annotated

		from fastapi import APIRouter, Depends, Path, Query, Response

		from federation_manager.api.errors import EWBI_ERROR_RESPONSES
		from federation_manager.api.security import get_bearer_token
		from federation_manager.application.authentication import PartnerAuthenticator
		from federation_manager.application.inbound_zones import InboundZoneService
		from federation_manager.contracts.ewbi import (
		ZoneRegisteredData,
		ZoneRegistrationRequestData,
		ZoneRegistrationResponseData,
		)
		from federation_manager.dependencies import get_inbound_zone_service, get_partner_authenticator
		from federation_manager.domain.ewbi import EWBI_BASE_PATH

		router = APIRouter(prefix=EWBI_BASE_PATH, tags=["AvailabilityZoneInfoSynchronization"])

		FederationContextIdPath = Annotated[str, Path(pattern=r"^[A-Za-z0-9][A-Za-z0-9-]*$")]
		Auth = Annotated[PartnerAuthenticator, Depends(get_partner_authenticator)]
		Service = Annotated[InboundZoneService, Depends(get_inbound_zone_service)]
		Token = Annotated[str, Depends(get_bearer_token)]


		@router.get(
		"/{federationContextId}/zones",
		operation_id="GetZoneData",
		response_model=ZoneRegisteredData,
		response_model_by_alias=True,
		responses=EWBI_ERROR_RESPONSES,
		)
		async def get_zone_data(
		federationContextId: FederationContextIdPath, # noqa: N803 - GSMA path template name
		auth: Auth,
		service: Service,
		token: Token,
		zoneId: Annotated[str \| None, Query()] = None, # noqa: N803 - GSMA query name
		) -> ZoneRegisteredData:
		partner = await auth.authenticate(token)
		return await service.zone_data(partner, federationContextId, zoneId)


		@router.post(
		"/{federationContextId}/zones",
		operation_id="ZoneSubscribe",
		response_model=ZoneRegistrationResponseData,
		response_model_by_alias=True,
		responses=EWBI_ERROR_RESPONSES,
		)
		async def subscribe_zones(
		federationContextId: FederationContextIdPath, # noqa: N803
		body: ZoneRegistrationRequestData,
		auth: Auth,
		service: Service,
		token: Token,
		) -> ZoneRegistrationResponseData:
		partner = await auth.authenticate(token)
		return await service.subscribe(partner, federationContextId, body)


		@router.get(
		"/{federationContextId}/zones/{zoneId}",
		operation_id="GetZoneDetails",
		response_model=ZoneRegisteredData,
		response_model_by_alias=True,
		responses=EWBI_ERROR_RESPONSES,
		)
		async def get_zone_details(
		federationContextId: FederationContextIdPath, # noqa: N803
		zoneId: str, # noqa: N803
		auth: Auth,
		service: Service,
		token: Token,
		) -> ZoneRegisteredData:
		partner = await auth.authenticate(token)
		return await service.zone_details(partner, federationContextId, zoneId)


		@router.delete(
		"/{federationContextId}/zones/{zoneId}",
		operation_id="ZoneUnsubscribe",
		responses=EWBI_ERROR_RESPONSES,
		)
		async def unsubscribe_zone(
		federationContextId: FederationContextIdPath, # noqa: N803
		zoneId: str, # noqa: N803
		auth: Auth,
		service: Service,
		token: Token,
		) -> Response:
		partner = await auth.authenticate(token)
		await service.unsubscribe(partner, federationContextId, zoneId)
		return Response(status_code=200)

src/federation_manager/api/errors.py

+32 −0

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

		@@ -358,3 +361,32 @@ def register_exception_handlers(app: FastAPI) -> None:
		"No federation partner offers this zone.",
		request.url.path,
		)

		@app.exception_handler(ZonesNotOffered)
		async def _zones_not_offered(request: Request, exc: ZonesNotOffered) -> JSONResponse:
		return problem(
		422,
		"zone-not-offered",
		"Zone Not Offered",
		f"These zones are not offered to you: {', '.join(exc.zone_ids)}.",
		request.url.path,
		)

		@app.exception_handler(ZoneIdRequired)
		async def _zone_id_required(request: Request, exc: ZoneIdRequired) -> JSONResponse:
		detail = (
		"No zones are offered to you."
		if exc.offered == 0
		else "Several zones are offered to you; pass zoneId to choose one."
		)
		return problem(422, "zone-id-required", "Zone Id Required", detail, request.url.path)

		@app.exception_handler(ZoneNotSubscribed)
		async def _zone_not_subscribed(request: Request, exc: ZoneNotSubscribed) -> JSONResponse:
		return problem(
		404,
		"zone-not-subscribed",
		"Zone Not Subscribed",
		"You have not subscribed to this zone.",
		request.url.path,
		)

mplement zone subscription flow

Merge request reports