feat: add CAMARA qod v1.1.0 schemas and router (5d47a4cb) · Commits · OOP / code / open-exposure-gateway

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

+921 −582

File changed.

Preview size limit exceeded, changes collapsed.

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

0 → 100644

+139 −0

Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v1_1_0/router.py: 139 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		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.v1_1_0.schemas import (
		CreateSession,
		ExtendSessionDuration,
		SessionInfo,
		)
		from open_exposure_gateway.application.services.quality_on_demand_service import (
		QualityOnDemandService,
		)
		from open_exposure_gateway.core.authorization import (
		require_role,
		require_scope,
		)
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		DownstreamServiceException,
		ForbiddenException,
		NotFoundException,
		NotImplementedException,
		UnauthorizedException,
		)
		from open_exposure_gateway.dependencies import CallerContext, get_caller_context, get_qod_service
		from open_exposure_gateway.schemas.common import ErrorInfo

		# CAMARA base path: the spec serves at {apiRoot}/quality-on-demand/v1 (wire version of v1.1.0).
		BASE_PATH = "/quality-on-demand/v1"

		router = APIRouter(
		prefix=BASE_PATH,
		dependencies=[Depends(require_role), Depends(require_scope)],
		)

		QoDService = Annotated[QualityOnDemandService, Depends(get_qod_service)]
		Caller = Annotated[CallerContext, Depends(get_caller_context)]

		# 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[422] = {"model": ErrorInfo, "description": "Unprocessable content"}
		_ERROR_RESPONSES[429] = {"model": ErrorInfo, "description": "Too many requests"}
		_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=["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, 404, 409, 422, 429, 500, 501, 503),
		)
		async def create_qod_session(
		request: CreateSession,
		service: QoDService,
		caller: Caller,
		x_correlator: XCorrelatorHeader = None,
		) -> Any:
		return await service.create_session(
		request=request, # type: ignore[arg-type]
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=caller.x_correlator,
		)


		@router.get(
		"/sessions/{sessionId}",
		tags=["QoS Sessions"],
		summary="Get QoS session information",
		response_model=SessionInfo,
		response_model_exclude_none=True,
		responses=_responses(400, 401, 403, 404, 429, 500, 503),
		)
		async def get_qod_session(
		sessionId: str,
		service: QoDService,
		caller: Caller,
		x_correlator: XCorrelatorHeader = None,
		) -> Any:
		return await service.get_session(
		session_id=sessionId,
		x_correlator=caller.x_correlator,
		)


		@router.delete(
		"/sessions/{sessionId}",
		tags=["QoS Sessions"],
		summary="Delete a QoS session",
		status_code=status.HTTP_204_NO_CONTENT,
		responses=_responses(400, 401, 403, 404, 429, 500, 503),
		)
		async def delete_qod_session(
		sessionId: str,
		service: QoDService,
		caller: Caller,
		x_correlator: XCorrelatorHeader = None,
		) -> Response:
		await service.delete_session(
		session_id=sessionId,
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=caller.x_correlator,
		)
		return Response(status_code=status.HTTP_204_NO_CONTENT)


		@router.post(
		"/sessions/{sessionId}/extend",
		tags=["QoS Sessions"],
		summary="Extend the duration of an active session",
		response_model=SessionInfo,
		response_model_exclude_none=True,
		responses=_responses(400, 401, 403, 404, 409, 429, 500, 501, 503),
		)
		async def extend_qod_session_duration(sessionId: str, request: ExtendSessionDuration) -> Any:
		raise NotImplementedException(message="Session duration extension is not implemented")

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

0 → 100644

+164 −0

Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v1_1_0/schemas.py: 164 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from enum import StrEnum
		from typing import Literal, Optional
		from uuid import UUID

		from pydantic import AwareDatetime, BaseModel, Field, field_validator, model_validator

		from open_exposure_gateway.api.camara.common import validated_ipv4, validated_ipv6

		SOURCE_API = "quality-on-demand"
		SOURCE_API_VERSION = "1.1.0"


		class QosStatus(StrEnum):
		REQUESTED = "REQUESTED"
		AVAILABLE = "AVAILABLE"
		UNAVAILABLE = "UNAVAILABLE"


		class StatusInfo(StrEnum):
		DURATION_EXPIRED = "DURATION_EXPIRED"
		NETWORK_TERMINATED = "NETWORK_TERMINATED"
		DELETE_REQUESTED = "DELETE_REQUESTED"


		class CredentialType(StrEnum):
		PLAIN = "PLAIN"
		ACCESSTOKEN = "ACCESSTOKEN"
		REFRESHTOKEN = "REFRESHTOKEN"


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

		@field_validator("publicAddress", "privateAddress")
		@classmethod
		def _valid_ipv4(cls, value: Optional[str]) -> Optional[str]:
		return validated_ipv4(value)

		@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] = Field(default=None, pattern=r"^\+[1-9][0-9]{4,14}$")
		networkAccessIdentifier: Optional[str] = None
		ipv4Address: Optional[DeviceIpv4Addr] = None
		ipv6Address: Optional[str] = None

		@field_validator("ipv6Address")
		@classmethod
		def _valid_ipv6(cls, value: Optional[str]) -> Optional[str]:
		return validated_ipv6(value)

		@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 DeviceResponse(Device):
		@model_validator(mode="after")
		def _require_exactly_one_identifier(self) -> "DeviceResponse":
		provided = [
		f
		for f in (
		self.phoneNumber,
		self.networkAccessIdentifier,
		self.ipv4Address,
		self.ipv6Address,
		)
		if f
		]
		if len(provided) != 1:
		raise ValueError("exactly 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", ge=0, le=65535)
		to: int = Field(ge=0, le=65535)

		model_config = {
		"populate_by_name": True,
		}


		class PortsSpec(BaseModel):
		ranges: Optional[list[PortRange]] = Field(default=None, min_length=1)
		ports: Optional[list[int]] = Field(default=None, min_length=1)

		@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 SinkCredential(BaseModel):
		credentialType: CredentialType
		accessToken: Optional[str] = None
		accessTokenExpiresUtc: Optional[AwareDatetime] = None
		accessTokenType: Optional[Literal["bearer"]] = None

		@model_validator(mode="after")
		def _require_access_token_fields(self) -> "SinkCredential":
		if self.credentialType is CredentialType.ACCESSTOKEN and (
		self.accessToken is None
		or self.accessTokenExpiresUtc is None
		or self.accessTokenType is None
		):
		raise ValueError(
		"accessToken, accessTokenExpiresUtc and accessTokenType are required "
		"for credentialType ACCESSTOKEN"
		)
		return self


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


		class CreateSession(BaseSessionInfo):
		device: Optional[Device] = None
		duration: int = Field(ge=1)


		class SessionInfo(BaseSessionInfo):
		device: Optional[DeviceResponse] = None
		sessionId: UUID
		duration: int = Field(ge=1)
		startedAt: Optional[AwareDatetime] = None
		expiresAt: Optional[AwareDatetime] = None
		qosStatus: QosStatus
		statusInfo: Optional[StatusInfo] = None


		class ExtendSessionDuration(BaseModel):
		requestedAdditionalDuration: int = Field(ge=1)