feat(fm): canonical GetFederationHealth replacefeat(fm): canonical... (022d21d0) · Commits · OOP / code / federation-manager

docs/OPG.04-v6.0-EWBI-Federation-API-v1.4.0-oop-profile.yaml

+2 −2

Original line number	Diff line number	Diff line
		# GENERATED by scripts/apply_overlay.py — do not edit.
		# base: OPG.04-v6.0-EWBI-Federation-API-v1.4.0.yaml (sha256 890801d61c148762897f18ce3b88823c0d486b1defdd04227f6a65f97c62fccf)
		# overlay: opg04-v1.4.0-oop-profile.overlay.yaml (version 0.1.0, 14 actions)
		# overlay: opg04-v1.4.0-oop-profile.overlay.yaml (version 0.1.0, 15 actions)
		openapi: 3.0.3
		info:
		version: 1.4.0
		@@ -319,7 +319,7 @@ components:
		- numOfAcceptedZones
		properties:
		federationStatus:
		$ref: '#/components/schemas/State'
		$ref: '#/components/schemas/Status'
		federationStartTime:
		$ref: '#/components/schemas/dateAndTimeZoneObject'
		numOfAcceptedZones:

docs/opg04-v1.4.0-oop-profile.overlay.yaml

+9 −0

Original line number	Diff line number	Diff line
		@@ -89,6 +89,15 @@ actions:
		type: object
		additionalProperties: true
		description: Result of the Service API processing, formatted according to mediaType and defined by the Service API specification.
		- target: $.components.schemas.FederationHealthInfo.properties.federationStatus
		description: >-
		federationStatus references State, the alarm-lifecycle object {alarmState}. OPG.04
		§3.1.1.12.2.12 Table 41 defines it as the federation Status enum, and the
		onPartnerStatusEvent callback in this same document already uses Status for the
		identically named field.
		update:
		$ref: '#/components/schemas/Status'

		- target: $.components.schemas.serviceAPIResponse.required
		description: >-
		The artifact requires both targetUserContext and apiResponse; OPG.04 §4.2.1.6.2.2 Table 197

src/federation_manager/adapters/database/federation_context_repo.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -44,6 +44,7 @@ class PostgresFederationContextRepo:
		federation_context_id=context.federation_context_id,
		status_callback_url=context.status_callback_url,
		status=context.status,
		created_at=context.created_at,
		)
		)
		await self._session.commit()
		@@ -58,6 +59,7 @@ class PostgresFederationContextRepo:
		direction=row.direction,
		federation_context_id=row.federation_context_id,
		status=row.status,
		created_at=row.created_at,
		agreement_id=row.agreement_id,
		status_callback_url=row.status_callback_url,
		)

src/federation_manager/api/errors.py

+51 −0

Original line number	Diff line number	Diff line
		from typing import Any

		from fastapi import FastAPI, Request
		from fastapi.responses import JSONResponse
		from pydantic.json_schema import models_json_schema

		from federation_manager.contracts.ewbi import InvalidParam, ProblemDetails
		from federation_manager.domain.errors import (
		AgreementExpired,
		AgreementViolation,
		AuthenticationFailed,
		FederationContextMissing,
		FederationContextUnknown,
		NoRouteMatched,
		PartnerEndpointConfigurationError,
		PartnerNotActive,
		@@ -19,6 +24,42 @@ from federation_manager.domain.errors import (

		_BEARER_CHALLENGE = {"WWW-Authenticate": 'Bearer scope="fed-mgmt"'}

		_PROBLEM_STATUSES = {
		400: "Bad request",
		401: "Unauthorized",
		404: "Not Found",
		409: "Conflict",
		422: "Unprocessable Entity",
		500: "Internal Server Error",
		503: "Service Unavailable",
		520: "Web Server Returned an Unknown Error",
		}

		# A "model" here would also emit application/json; a $ref needs register_problem_schemas().
		EWBI_ERROR_RESPONSES: dict[int \| str, dict[str, Any]] = {
		**{
		status: {
		"description": description,
		"content": {
		"application/problem+json": {
		"schema": {"$ref": "#/components/schemas/ProblemDetails"}
		}
		},
		}
		for status, description in _PROBLEM_STATUSES.items()
		},
		"default": {"description": "Generic Error"},
		}


		def register_problem_schemas(schema: dict[str, Any]) -> dict[str, Any]:
		_, defs = models_json_schema(
		[(ProblemDetails, "validation"), (InvalidParam, "validation")],
		ref_template="#/components/schemas/{model}",
		)
		schema.setdefault("components", {}).setdefault("schemas", {}).update(defs["$defs"])
		return schema


		def problem(
		status: int,
		@@ -129,6 +170,16 @@ def register_exception_handlers(app: FastAPI) -> None:
		request.url.path,
		)

		@app.exception_handler(FederationContextUnknown)
		async def _context_unknown(request: Request, exc: FederationContextUnknown) -> JSONResponse:
		return problem(
		404,
		"federation-context-unknown",
		"Unknown Federation Context",
		"No federation context with this identifier exists for the calling partner.",
		request.url.path,
		)

		@app.exception_handler(FederationContextMissing)
		async def _context_missing(request: Request, exc: FederationContextMissing) -> JSONResponse:
		return problem(

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

+43 −8

Original line number	Diff line number	Diff line
		from typing import Annotated

		from fastapi import APIRouter, Depends
		from fastapi import APIRouter, Depends, Path

		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.dependencies import get_partner_authenticator
		from federation_manager.contracts.ewbi import (
		FederationHealthInfo,
		FederationHealthResponse,
		FederationStatus,
		)
		from federation_manager.dependencies import get_federation_context_repo, get_partner_authenticator
		from federation_manager.domain.errors import FederationContextUnknown
		from federation_manager.domain.ewbi import EWBI_BASE_PATH
		from federation_manager.domain.ports import FederationContextRepositoryPort

		router = APIRouter(prefix="/ewbi/v1/management", tags=["ewbi-management"])
		router = APIRouter(prefix=EWBI_BASE_PATH, tags=["FederationManagement"])

		FederationContextIdPath = Annotated[str, Path(pattern=r"^[A-Za-z0-9][A-Za-z0-9-]*$")]

		@router.post("/heartbeat")
		async def heartbeat(
		_WIRE_STATUS: dict[str, FederationStatus] = {
		"available": "AVAILABLE",
		"locked": "LOCKED",
		"not_available": "NOT_AVAILABLE",
		"temporary_failure": "TEMPORARY_FAILURE",
		"failed": "FAILED",
		}


		@router.get(
		"/{federationContextId}/health",
		operation_id="GetFederationHealth",
		responses=EWBI_ERROR_RESPONSES,
		)
		async def get_federation_health(
		federationContextId: FederationContextIdPath, # noqa: N803 - GSMA path template name
		auth: Annotated[PartnerAuthenticator, Depends(get_partner_authenticator)],
		contexts: Annotated[FederationContextRepositoryPort, Depends(get_federation_context_repo)],
		token: Annotated[str, Depends(get_bearer_token)],
		) -> dict[str, str]:
		await auth.authenticate(token)
		return {"status": "ALIVE"}
		) -> FederationHealthResponse:
		partner = await auth.authenticate(token)
		context = await contexts.find_inbound(partner.id, federationContextId)
		if context is None:
		raise FederationContextUnknown(partner.id)
		return FederationHealthResponse(
		federation_health_status=FederationHealthInfo(
		federation_status=_WIRE_STATUS.get(context.status, "NOT_AVAILABLE"),
		federation_start_time=context.created_at,
		# no catalogue sync yet
		num_of_accepted_zones="0",
		)
		)