fix: align session request/response schemas with CAMARA spec (05301195) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/http/srm_client.py

+5 −5

Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@ import httpx
		import structlog

		from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (
		QoDSessionResponse,
		SessionInfo,
		)
		from open_exposure_gateway.core.config import get_settings
		from open_exposure_gateway.core.exceptions import (
		@@ -128,19 +128,19 @@ class SRMClient:
		self,
		payload: dict[str, Any],
		x_correlator: str \| None = None,
		) -> QoDSessionResponse:
		) -> SessionInfo:
		headers = {"x-correlator": x_correlator} if x_correlator else None
		data = await self._request("POST", "/sessions", json=payload, headers=headers)
		return QoDSessionResponse.model_validate(data)
		return SessionInfo.model_validate(data)

		async def get_qod_session(
		self,
		session_id: str,
		x_correlator: str \| None = None,
		) -> QoDSessionResponse:
		) -> SessionInfo:
		headers = {"x-correlator": x_correlator} if x_correlator else None
		data = await self._request("GET", f"/sessions/{session_id}", headers=headers)
		return QoDSessionResponse.model_validate(data)
		return SessionInfo.model_validate(data)

		async def delete_qod_session(
		self,

src/open_exposure_gateway/api/camara/quality_on_demand/API_definitions/qod-api.yaml

0 → 100644

+1234 −0

File added.

Preview size limit exceeded, changes collapsed.

src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/router.py

+50 −12

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

		from fastapi import APIRouter, Depends, Response, status

		from open_exposure_gateway.api.camara.common import XCorrelatorHeader
		from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (
		QoDSessionRequest,
		QoDSessionResponse,
		CreateSession,
		SessionInfo,
		)
		from open_exposure_gateway.application.services.quality_on_demand_service import (
		QualityOnDemandService,
		)
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		DownstreamServiceException,
		ForbiddenException,
		NotFoundException,
		NotImplementedException,
		UnauthorizedException,
		)
		from open_exposure_gateway.dependencies import get_qod_service
		from open_exposure_gateway.schemas.common import ErrorInfo

		# CAMARA base path: the spec serves at {apiRoot}/qod/v0 (wire version of v0.10.1).
		BASE_PATH = "/qod/v0"
		@@ -19,18 +29,42 @@ router = APIRouter(prefix=BASE_PATH)

		QoDService = Annotated[QualityOnDemandService, Depends(get_qod_service)]

		# OpenAPI response docs derived from core.exceptions' status_code/message
		# defaults, so codes aren't re-listed. 500 has no dedicated exception class
		# (it's the unhandled-exception catch-all in error_handlers.py).
		_ERROR_RESPONSES: dict[int \| str, dict[str, Any]] = {
		exc_cls().status_code: {"model": ErrorInfo, "description": exc_cls().message}
		for exc_cls in (
		BadRequestException,
		UnauthorizedException,
		ForbiddenException,
		NotFoundException,
		ConflictException,
		NotImplementedException,
		DownstreamServiceException,
		)
		}
		_ERROR_RESPONSES[500] = {"model": ErrorInfo, "description": "Internal server error"}


		def _responses(*codes: int) -> dict[int \| str, dict[str, Any]]:
		return {code: _ERROR_RESPONSES[code] for code in codes}


		@router.post(
		"/sessions",
		tags=["Quality on Demand Functions"],
		summary="Creates a new QoD Session",
		tags=["QoS Sessions"],
		summary="Creates a new session",
		response_model=SessionInfo,
		response_model_exclude_none=True,
		status_code=status.HTTP_201_CREATED,
		responses=_responses(400, 401, 403, 409, 500, 501, 503),
		)
		async def create_qod_session(
		request: QoDSessionRequest,
		request: CreateSession,
		service: QoDService,
		x_correlator: XCorrelatorHeader = None,
		) -> QoDSessionResponse:
		) -> Any:
		return await service.create_session(
		request=request,
		x_correlator=x_correlator,
		@@ -39,14 +73,17 @@ async def create_qod_session(

		@router.get(
		"/sessions/{sessionId}",
		tags=["Quality on Demand Functions"],
		summary="Retrieve details of a QoD Session",
		tags=["QoS Sessions"],
		summary="Get QoS session information",
		response_model=SessionInfo,
		response_model_exclude_none=True,
		responses=_responses(400, 401, 403, 404, 500, 503),
		)
		async def get_qod_session(
		sessionId: str,
		service: QoDService,
		x_correlator: XCorrelatorHeader = None,
		) -> QoDSessionResponse:
		) -> Any:
		return await service.get_session(
		session_id=sessionId,
		x_correlator=x_correlator,
		@@ -55,9 +92,10 @@ async def get_qod_session(

		@router.delete(
		"/sessions/{sessionId}",
		tags=["Quality on Demand Functions"],
		summary="Remove QoD Session",
		tags=["QoS Sessions"],
		summary="Delete a QoS session",
		status_code=status.HTTP_204_NO_CONTENT,
		responses=_responses(400, 401, 403, 404, 500, 503),
		)
		async def delete_qod_session(
		sessionId: str,

src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/schemas.py

+54 −17

Original line number	Diff line number	Diff line
		from enum import StrEnum
		from typing import Optional
		from typing import Literal, Optional
		from uuid import UUID

		from pydantic import BaseModel, Field
		from pydantic import BaseModel, Field, model_validator


		class QosStatus(StrEnum):
		@@ -11,27 +11,49 @@ class QosStatus(StrEnum):
		UNAVAILABLE = "UNAVAILABLE"


		class Ipv4Address(BaseModel):
		class DeviceIpv4Addr(BaseModel):
		publicAddress: Optional[str] = None
		privateAddress: Optional[str] = None
		publicPort: Optional[int] = Field(default=None, ge=0, le=65535)

		@model_validator(mode="after")
		def _require_public_plus_one(self) -> "DeviceIpv4Addr":
		if self.publicAddress is None:
		raise ValueError("publicAddress is required")
		if self.privateAddress is None and self.publicPort is None:
		raise ValueError("at least one of privateAddress or publicPort is required")
		return self


		class Device(BaseModel):
		phoneNumber: Optional[str] = None
		phoneNumber: Optional[str] = Field(default=None, pattern=r"^\+?[0-9]{5,15}$")
		networkAccessIdentifier: Optional[str] = None
		ipv4Address: Optional[Ipv4Address] = None
		ipv4Address: Optional[DeviceIpv4Addr] = None
		ipv6Address: Optional[str] = None

		@model_validator(mode="after")
		def _require_at_least_one_identifier(self) -> "Device":
		if not any(
		(self.phoneNumber, self.networkAccessIdentifier, self.ipv4Address, self.ipv6Address)
		):
		raise ValueError("at least one device identifier must be provided")
		return self


		class ApplicationServer(BaseModel):
		ipv4Address: Optional[str] = None
		ipv6Address: Optional[str] = None

		@model_validator(mode="after")
		def _require_at_least_one_address(self) -> "ApplicationServer":
		if not any((self.ipv4Address, self.ipv6Address)):
		raise ValueError("at least one of ipv4Address or ipv6Address must be provided")
		return self


		class PortRange(BaseModel):
		from_: int = Field(alias="from")
		to: int
		from_: int = Field(alias="from", ge=0, le=65535)
		to: int = Field(ge=0, le=65535)

		model_config = {
		"populate_by_name": True,
		@@ -42,25 +64,40 @@ class PortsSpec(BaseModel):
		ranges: Optional[list[PortRange]] = None
		ports: Optional[list[int]] = None

		@model_validator(mode="after")
		def _require_ranges_or_ports(self) -> "PortsSpec":
		if not self.ranges and not self.ports:
		raise ValueError("at least one of ranges or ports must be provided")
		return self


		class Webhook(BaseModel):
		notificationUrl: str
		notificationAuthToken: Optional[str] = Field(default=None, min_length=20, max_length=256)


		class Message(BaseModel):
		severity: Literal["INFO", "WARNING"]
		description: str

		class QoDSessionRequest(BaseModel):

		class BaseSessionInfo(BaseModel):
		device: Device
		applicationServer: ApplicationServer
		qosProfile: str
		devicePorts: Optional[PortsSpec] = None
		applicationServerPorts: Optional[PortsSpec] = None
		sink: Optional[str] = None
		qosProfile: str = Field(min_length=3, max_length=256, pattern=r"^[a-zA-Z0-9_.-]+$")
		webhook: Optional[Webhook] = None


		class CreateSession(BaseSessionInfo):
		duration: int = Field(default=86400, ge=1, le=86400)


		class QoDSessionResponse(BaseModel):
		class SessionInfo(BaseSessionInfo):
		sessionId: UUID
		device: Device
		applicationServer: ApplicationServer
		qosProfile: str
		duration: int
		duration: int = Field(ge=1)
		startedAt: int
		expiresAt: int
		qosStatus: QosStatus
		devicePorts: Optional[PortsSpec] = None
		applicationServerPorts: Optional[PortsSpec] = None
		messages: Optional[list[Message]] = None

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+5 −5

Original line number	Diff line number	Diff line
		from typing import Optional

		from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (
		QoDSessionRequest,
		QoDSessionResponse,
		CreateSession,
		SessionInfo,
		)
		from open_exposure_gateway.ports.srm_port import SRMClientPort

		@@ -13,9 +13,9 @@ class QualityOnDemandService:

		async def create_session(
		self,
		request: QoDSessionRequest,
		request: CreateSession,
		x_correlator: Optional[str] = None,
		) -> QoDSessionResponse:
		) -> SessionInfo:
		return await self.srm_client.create_qod_session(
		payload=request.model_dump(mode="json"),
		x_correlator=x_correlator,
		@@ -25,7 +25,7 @@ class QualityOnDemandService:
		self,
		session_id: str,
		x_correlator: Optional[str] = None,
		) -> QoDSessionResponse:
		) -> SessionInfo:
		return await self.srm_client.get_qod_session(
		session_id=session_id,
		x_correlator=x_correlator,

Admin message