Commit 042a4bc4 authored by Sergio Gimenez's avatar Sergio Gimenez
Browse files

feat(fm): deploy on a partner: onboard over the EWBI, then InstallApp (ADR-0061)

parent 455991f4

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/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")

src/federation_manager/api/errors.py

+55 −0
Changes for src/federation_manager/api/errors.py: 55 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -15,8 +15,12 @@ from federation_manager.domain.errors import ( 
    FederationContextUnknown,

    FederationIdentityMismatch,

    IdempotencyKeyReused,

    InstanceUnknown,

    NetworkQueryNotApplicable,

    NoRouteMatched,

    OnboardingConflict,

    OnboardingNotSupported,

    OnboardingResourceUnknown,

    PartnerCredentialsIncomplete,

    PartnerEndpointConfigurationError,

    PartnerNotActive,
@@ -28,6 +32,7 @@ from federation_manager.domain.errors import ( 
    PartnerTokenConfigurationError,

    PartnerTokenRequestFailed,

    PartnerUnknown,

    ServiceSpecificationUnknown,

    SessionUnknown,

    SrmQueryFailed,

    UnsupportedServiceApi,
@@ -372,6 +377,56 @@ def register_exception_handlers(app: FastAPI) -> None: 
            request.url.path,

        )



    @app.exception_handler(InstanceUnknown)

    async def _instance_unknown(request: Request, exc: InstanceUnknown) -> JSONResponse:

        return problem(

            404,

            "instance-unknown",

            "Unknown Application Instance",

            "No application instance with this identifier was installed by us at your platform.",

            request.url.path,

        )



    @app.exception_handler(ServiceSpecificationUnknown)

    async def _specification_unknown(

        request: Request, exc: ServiceSpecificationUnknown

    ) -> JSONResponse:

        return problem(

            404,

            "service-specification-unknown",

            "Service Specification Unknown",

            f"No service specification {exc.specification_id} exists.",

            request.url.path,

        )



    @app.exception_handler(OnboardingNotSupported)

    async def _onboarding_not_supported(

        request: Request, exc: OnboardingNotSupported

    ) -> JSONResponse:

        return problem(422, "not-supported", "Not Supported", exc.reason, request.url.path)



    @app.exception_handler(OnboardingResourceUnknown)

    async def _onboarding_resource_unknown(

        request: Request, exc: OnboardingResourceUnknown

    ) -> JSONResponse:

        return problem(

            422,

            f"{exc.kind}-unknown",

            "Unknown Reference",

            f"These {exc.kind}s were not uploaded: {', '.join(exc.ids)}.",

            request.url.path,

        )



    @app.exception_handler(OnboardingConflict)

    async def _onboarding_conflict(request: Request, exc: OnboardingConflict) -> JSONResponse:

        return problem(

            409,

            f"{exc.kind}-exists",

            "Already Exists",

            f"The {exc.kind} {exc.resource_id} already exists.",

            request.url.path,

        )



    @app.exception_handler(ZoneIdRequired)

    async def _zone_id_required(request: Request, exc: ZoneIdRequired) -> JSONResponse:

        detail = (