Add inbound QoD session support (UC-FM-04) (!16) · Merge requests · OOP / code / federation-manager

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

+46 −4

Changes for src/federation_manager/api/ewbi/v1/service.py: 46 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		from typing import Annotated
		from uuid import UUID

		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.capabilities import QUALITY_ON_DEMAND, InboundCapabilityService
		from federation_manager.application.queries import InboundQueryService
		from federation_manager.contracts.ewbi import ApiForwardingRequest, ServiceApiResponse
		from federation_manager.dependencies import get_inbound_query_service
		from federation_manager.contracts.ewbi import (
		ApiForwardingRequest,
		ServiceApiResponse,
		SessionContext,
		)
		from federation_manager.dependencies import (
		get_inbound_capability_service,
		get_inbound_query_service,
		)
		from federation_manager.domain.ewbi import EWBI_BASE_PATH

		router = APIRouter(prefix=EWBI_BASE_PATH, tags=["ServiceAPIManagement"])
		@@ -23,7 +32,40 @@ async def api_forwarding(
		federationContextId: FederationContextIdPath, # noqa: N803 - GSMA path template name
		serviceAPINameVal: str, # noqa: N803 - GSMA path template name
		body: ApiForwardingRequest,
		service: Annotated[InboundQueryService, Depends(get_inbound_query_service)],
		queries: Annotated[InboundQueryService, Depends(get_inbound_query_service)],
		capabilities: Annotated[InboundCapabilityService, Depends(get_inbound_capability_service)],
		token: Annotated[str, Depends(get_bearer_token)],
		) -> ServiceApiResponse:
		return await service.forward(token, federationContextId, serviceAPINameVal, body)
		if serviceAPINameVal == QUALITY_ON_DEMAND:
		return await capabilities.open_session(token, federationContextId, body)
		return await queries.forward(token, federationContextId, serviceAPINameVal, body)


		@router.get(
		"/{federationContextId}/apiservice/connid/{connectID}/custid/{customerID}",
		operation_id="GetServiceAPISessionInfo",
		responses=EWBI_ERROR_RESPONSES,
		)
		async def get_service_api_session_info(
		federationContextId: FederationContextIdPath, # noqa: N803 - GSMA path template name
		connectID: str, # noqa: N803 - GSMA path template name
		customerID: UUID, # noqa: N803 - GSMA path template name
		service: Annotated[InboundCapabilityService, Depends(get_inbound_capability_service)],
		token: Annotated[str, Depends(get_bearer_token)],
		) -> SessionContext:
		return await service.session_info(token, federationContextId, connectID, customerID)


		@router.delete(
		"/{federationContextId}/apiservice/connid/{connectID}/custid/{customerID}",
		operation_id="RemoveServiceAPISession",
		responses=EWBI_ERROR_RESPONSES,
		)
		async def remove_service_api_session(
		federationContextId: FederationContextIdPath, # noqa: N803 - GSMA path template name
		connectID: str, # noqa: N803 - GSMA path template name
		customerID: UUID, # noqa: N803 - GSMA path template name
		service: Annotated[InboundCapabilityService, Depends(get_inbound_capability_service)],
		token: Annotated[str, Depends(get_bearer_token)],
		) -> SessionContext:
		return await service.close_session(token, federationContextId, connectID, customerID)

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
		@@ -23,6 +23,7 @@ from federation_manager.domain.errors import (
		PartnerTokenConfigurationError,
		PartnerTokenRequestFailed,
		PartnerUnknown,
		SessionUnknown,
		SrmQueryFailed,
		UnsupportedServiceApi,
		problem_type,
		@@ -240,6 +241,16 @@ def register_exception_handlers(app: FastAPI) -> None:
		request.url.path,
		)

		@app.exception_handler(SessionUnknown)
		async def _session_unknown(request: Request, exc: SessionUnknown) -> JSONResponse:
		return problem(
		404,
		"session-unknown",
		"Unknown Service API Session",
		"No Service API session with this identifier exists for the calling partner.",
		request.url.path,
		)

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

src/federation_manager/application/capabilities.py

0 → 100644

+249 −0

Changes for src/federation_manager/application/capabilities.py: 249 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from collections.abc import Callable
		from datetime import datetime, timezone
		from uuid import UUID, uuid4

		from federation_manager.application.authentication import PartnerAuthenticator
		from federation_manager.application.authorization import AgreementChecker
		from federation_manager.contracts.ewbi import (
		ApiForwardingRequest,
		ExpiryInterval,
		ServiceApiResponse,
		SessionContext,
		TargetUserContext,
		)
		from federation_manager.contracts.srm import (
		SUBJECT_CAPABILITY_ACTIVATE,
		SUBJECT_CAPABILITY_DEACTIVATE,
		CapabilityParameters,
		CapabilityTarget,
		DeviceTarget,
		NetworkCapabilityDeactivatePayloadV1,
		NetworkCapabilityPayloadV1,
		SourceSpecificationV1,
		SrmNetworkCapabilityActivateV1,
		SrmNetworkCapabilityDeactivateV1,
		)
		from federation_manager.domain.errors import (
		AgreementViolation,
		FederationContextUnknown,
		SessionUnknown,
		)
		from federation_manager.domain.models import (
		FederationContext,
		FederationTransaction,
		PartnerOP,
		)
		from federation_manager.domain.ports import (
		DataBusPublisherPort,
		FederationContextRepositoryPort,
		TransactionRepositoryPort,
		)

		INBOUND = "inbound"
		QUALITY_ON_DEMAND = "QualityOnDemand"
		API_TYPE = "qod-session"
		DEACTIVATE_API_TYPE = "qod-session-remove"
		CAPABILITY_TYPE = "qod_session"
		DEFAULT_DURATION_SECONDS = 3600
		# expiryInterval caps numHours at 23, so a session longer than this cannot be expressed back
		MAX_DURATION_SECONDS = 86399
		OPG_SOURCE = SourceSpecificationV1(family="gsma_opg", api="APIForwarding", version="1.4.0")


		def _utcnow() -> datetime:
		return datetime.now(timezone.utc)


		class InboundCapabilityService:
		def __init__(
		self,
		authenticator: PartnerAuthenticator,
		contexts: FederationContextRepositoryPort,
		agreements: AgreementChecker,
		transactions: TransactionRepositoryPort,
		publisher: DataBusPublisherPort,
		*,
		clock: Callable[[], datetime] = _utcnow,
		id_factory: Callable[[], UUID] = uuid4,
		) -> None:
		self._authenticator = authenticator
		self._contexts = contexts
		self._agreements = agreements
		self._transactions = transactions
		self._publisher = publisher
		self._clock = clock
		self._new_id = id_factory

		async def open_session(
		self, token: str, federation_context_id: str, request: ApiForwardingRequest
		) -> ServiceApiResponse:
		partner, context = await self._partner_context(token, federation_context_id)
		agreement = await self._agreements.require(partner, API_TYPE)
		specification_id = agreement.resolve_api_spec(API_TYPE)
		if specification_id is None:
		raise AgreementViolation

		content = request.service_api_body.api_content
		session_id = self._new_id()
		operation_id = self._new_id()
		correlation_id = self._new_id()
		duration = _duration_seconds(content)

		transaction = FederationTransaction(
		id=self._new_id(),
		partner_op_id=partner.id,
		agreement_id=agreement.id,
		federation_context_row_id=context.id,
		direction=INBOUND,
		operation_id=operation_id,
		correlation_id=correlation_id,
		api_type=API_TYPE,
		status="pending",
		external_resource_id=session_id.hex,
		external_txn_id=request.txn_identifier,
		request_summary={
		"customer_id": str(request.customer_id),
		"duration_seconds": duration,
		"federation_context_id": federation_context_id,
		"service_specification_id": str(specification_id),
		},
		started_at=self._clock(),
		)
		await self._transactions.add(transaction)

		command = SrmNetworkCapabilityActivateV1(
		operation_id=operation_id,
		correlation_id=str(correlation_id),
		requested_at=self._clock(),
		app_provider_id=str(partner.id),
		federation_partner_ref=partner.mcc_mnc,
		source="federation",
		service_instance_id=session_id,
		service_specification_id=specification_id,
		network_capability=NetworkCapabilityPayloadV1(
		capability_type=CAPABILITY_TYPE,
		target=CapabilityTarget(device=_device(content)),
		parameters=CapabilityParameters(duration_seconds=duration),
		profile_ref=_text(content.get("qosProfile")),
		source_spec=OPG_SOURCE,
		),
		)
		await self._publisher.publish(SUBJECT_CAPABILITY_ACTIVATE, command.model_dump(mode="json"))
		await self._transactions.mark_in_progress(transaction.id)

		return ServiceApiResponse(
		customer_id=request.customer_id,
		txn_identifier=request.txn_identifier,
		target_user_context=TargetUserContext(
		connect_id=session_id.hex, expiry_duration=_expiry(duration)
		),
		)

		async def session_info(
		self, token: str, federation_context_id: str, connect_id: str, customer_id: UUID
		) -> SessionContext:
		_, session = await self._session(token, federation_context_id, connect_id, customer_id)
		duration = int(str(session.request_summary.get("duration_seconds", 0)))
		return SessionContext(connect_id=connect_id, expiry_duration=_expiry(duration))

		async def close_session(
		self, token: str, federation_context_id: str, connect_id: str, customer_id: UUID
		) -> SessionContext:
		partner, session = await self._session(
		token, federation_context_id, connect_id, customer_id
		)
		duration = int(str(session.request_summary.get("duration_seconds", 0)))
		closing = SessionContext(connect_id=connect_id, expiry_duration=_expiry(duration))

		already = await self._transactions.find_by_external_resource_id(
		partner.id, DEACTIVATE_API_TYPE, connect_id
		)
		if already is not None:
		return closing

		operation_id = self._new_id()
		correlation_id = self._new_id()
		specification_id = session.request_summary.get("service_specification_id")
		await self._transactions.add(
		FederationTransaction(
		id=self._new_id(),
		partner_op_id=partner.id,
		agreement_id=session.agreement_id,
		federation_context_row_id=session.federation_context_row_id,
		direction=INBOUND,
		operation_id=operation_id,
		correlation_id=correlation_id,
		api_type=DEACTIVATE_API_TYPE,
		status="pending",
		external_resource_id=connect_id,
		request_summary={"customer_id": str(customer_id)},
		started_at=self._clock(),
		)
		)
		command = SrmNetworkCapabilityDeactivateV1(
		operation_id=operation_id,
		correlation_id=str(correlation_id),
		requested_at=self._clock(),
		app_provider_id=str(partner.id),
		federation_partner_ref=partner.mcc_mnc,
		source="federation",
		service_specification_id=UUID(str(specification_id)) if specification_id else None,
		network_capability=NetworkCapabilityDeactivatePayloadV1(
		capability_type=CAPABILITY_TYPE, service_instance_id=UUID(connect_id)
		),
		)
		await self._publisher.publish(
		SUBJECT_CAPABILITY_DEACTIVATE, command.model_dump(mode="json")
		)
		return closing

		async def _partner_context(
		self, token: str, federation_context_id: str
		) -> tuple[PartnerOP, FederationContext]:
		partner = await self._authenticator.authenticate(token)
		context = await self._contexts.find_inbound(partner.id, federation_context_id)
		if context is None or context.is_terminated():
		raise FederationContextUnknown(partner.id)
		return partner, context

		async def _session(
		self, token: str, federation_context_id: str, connect_id: str, customer_id: UUID
		) -> tuple[PartnerOP, FederationTransaction]:
		partner, _ = await self._partner_context(token, federation_context_id)
		session = await self._transactions.find_by_external_resource_id(
		partner.id, API_TYPE, connect_id
		)
		if session is None or session.request_summary.get("customer_id") != str(customer_id):
		raise SessionUnknown(connect_id)
		return partner, session


		def _device(content: dict[str, object]) -> DeviceTarget:
		device = content.get("device")
		if not isinstance(device, dict):
		raise AgreementViolation
		target = DeviceTarget(
		phone_number=_text(device.get("phoneNumber")),
		ipv6=_text(device.get("ipv6Address")),
		network_access_id=_text(device.get("networkAccessIdentifier")),
		)
		if not any((target.phone_number, target.ipv6, target.network_access_id)):
		raise AgreementViolation
		return target


		def _duration_seconds(content: dict[str, object]) -> int:
		duration = content.get("duration")
		if not isinstance(duration, int) or duration <= 0:
		return DEFAULT_DURATION_SECONDS
		return min(duration, MAX_DURATION_SECONDS)


		def _expiry(seconds: int) -> ExpiryInterval:
		return ExpiryInterval(
		num_hours=seconds // 3600, num_mins=(seconds % 3600) // 60, num_secs=seconds % 60
		)


		def _text(value: object) -> str \| None:
		return value if isinstance(value, str) and value else None

src/federation_manager/contracts/ewbi.py

+9 −0

Changes for src/federation_manager/contracts/ewbi.py: 9 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -258,3 +258,12 @@ class InstanceStatusCallback(BaseModel):
		app_instance_info: AppInstanceInfo = Field(
		validation_alias="appInstanceInfo", serialization_alias="appInstanceInfo"
		)


		class SessionContext(BaseModel):
		model_config = ConfigDict(populate_by_name=True)

		connect_id: str = Field(validation_alias="connectID", serialization_alias="connectID")
		expiry_duration: ExpiryInterval = Field(
		validation_alias="expiryDuration", serialization_alias="expiryDuration"
		)

src/federation_manager/contracts/srm.py

+27 −0

Changes for src/federation_manager/contracts/srm.py: 27 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -13,6 +13,8 @@ EVENT_STREAM = "OOP_EVENTS"
		SUBJECT_OPERATION_COMPLETED = "event.srm.operation.completed"
		SUBJECT_DEPLOY = "command.srm.service.deploy"
		SUBJECT_TERMINATE = "command.srm.service.terminate"
		SUBJECT_CAPABILITY_ACTIVATE = "command.srm.network.capability.activate"
		SUBJECT_CAPABILITY_DEACTIVATE = "command.srm.network.capability.deactivate"


		class CommandEnvelopeV1(BaseModel):
		@@ -103,6 +105,7 @@ class CapabilityTarget(BaseModel):

		class CapabilityParameters(BaseModel):
		max_age_seconds: int \| None = None
		duration_seconds: int \| None = None


		class SourceSpecificationV1(BaseModel):
		@@ -126,3 +129,27 @@ class LocationQueryResponseV1(BaseModel):
		last_location_time: datetime
		area: dict[str, Any]
		srm_resolved_identifier: str \| None = None


		class NetworkCapabilityPayloadV1(BaseModel):
		capability_type: str
		target: CapabilityTarget
		parameters: CapabilityParameters
		profile_ref: str \| None = None
		source_spec: SourceSpecificationV1 \| None = None


		class SrmNetworkCapabilityActivateV1(CommandEnvelopeV1):
		service_instance_id: UUID
		service_specification_id: UUID
		network_capability: NetworkCapabilityPayloadV1


		class NetworkCapabilityDeactivatePayloadV1(BaseModel):
		capability_type: str
		service_instance_id: UUID


		class SrmNetworkCapabilityDeactivateV1(CommandEnvelopeV1):
		network_capability: NetworkCapabilityDeactivatePayloadV1
		service_specification_id: UUID \| None = None

Add inbound QoD session support (UC-FM-04)

What a partner can now do

Known Limitations

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

src/federation_manager/api/errors.py

src/federation_manager/application/capabilities.py

src/federation_manager/contracts/ewbi.py

src/federation_manager/contracts/srm.py

Add inbound QoD session support (UC-FM-04)

What a partner can now do

Known Limitations

Merge request reports