feat: implement location retrieval api (!25) · Merge requests · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/http/srm_client.py

+102 −1

Changes for src/open_exposure_gateway/adapters/http/srm_client.py: 102 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -3,11 +3,14 @@ from uuid import UUID

		import httpx
		import structlog
		from pydantic import ValidationError

		from open_exposure_gateway.core.config import get_settings
		from open_exposure_gateway.core.exceptions import (
		DownstreamServiceException,
		ErrorCode,
		NotFoundException,
		UnprocessableEntityException,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		SRMCatalogPayload,
		@@ -15,10 +18,56 @@ from open_exposure_gateway.domain.edge_application_management import (
		SRMServiceInstance,
		SRMZone,
		)
		from open_exposure_gateway.domain.location_retrieval import (
		SRMLocationQuery,
		SRMLocationResult,
		)
		from open_exposure_gateway.domain.quality_on_demand import SRMNetworkCapability

		logger = structlog.get_logger(__name__)

		# SRM's RFC 7807 problem `type` -> the CAMARA code and the message the vendored spec pairs
		# with it (srm/interface-contract.md §E.2).
		# The message is fixed here rather than passed through from `detail` because this body crosses
		# the trust boundary to the app provider; `detail` is logged instead.
		_PROBLEM_NS = "https://etsi.org/sdg/oop/problems/"
		_QUERY_PROBLEMS: dict[str, tuple[ErrorCode, str]] = {
		f"{_PROBLEM_NS}unable-to-locate": (
		ErrorCode.LOCATION_RETRIEVAL_UNABLE_TO_LOCATE,
		"The network is unable to locate the device",
		),
		f"{_PROBLEM_NS}unable-to-fulfill-max-age": (
		ErrorCode.LOCATION_RETRIEVAL_UNABLE_TO_FULFILL_MAX_AGE,
		"Unable to provide expected freshness for location",
		),
		f"{_PROBLEM_NS}unable-to-fulfill-max-surface": (
		ErrorCode.LOCATION_RETRIEVAL_UNABLE_TO_FULFILL_MAX_SURFACE,
		"Unable to provide accurate acceptable surface for location",
		),
		f"{_PROBLEM_NS}unsupported-identifier": (
		ErrorCode.UNSUPPORTED_IDENTIFIER,
		"The identifier provided is not supported.",
		),
		f"{_PROBLEM_NS}service-not-applicable": (
		ErrorCode.SERVICE_NOT_APPLICABLE,
		"The service is not available for the provided identifier.",
		),
		}

		_UNKNOWN_QUERY_PROBLEM = _QUERY_PROBLEMS[f"{_PROBLEM_NS}service-not-applicable"]

		_PROBLEM_MEDIA_TYPE = "application/problem+json"


		def _problem_document(response: httpx.Response) -> dict[str, Any] \| None:
		if _PROBLEM_MEDIA_TYPE not in response.headers.get("content-type", ""):
		return None
		try:
		body = response.json()
		except ValueError: # JSONDecodeError -- truncated or mislabelled body
		return None
		return body if isinstance(body, dict) else None


		class SRMClient:
		def __init__(self) -> None:
		@@ -33,6 +82,7 @@ class SRMClient:
		json: dict[str, Any] \| None = None,
		params: dict[str, Any] \| None = None,
		headers: dict[str, str] \| None = None,
		problems: dict[str, tuple[ErrorCode, str]] \| None = None,
		) -> Any:
		url = f"{self.base_url}{path}"
		log = logger.bind(
		@@ -51,10 +101,26 @@ class SRMClient:
		headers=headers,
		)

		if response.status_code == 404:
		# Only the query calls translate SRM problems into CAMARA codes; every other call
		# keeps the plain 404 handling it has always had.
		problem = _problem_document(response) if problems is not None else None

		if response.status_code == 404 and (problems is None or problem is not None):
		log.warning("SRM resource not found", url=url)
		raise NotFoundException(message="Resource not found")

		if response.status_code == 422 and problems is not None and problem is not None:
		problem_type = problem.get("type")
		key = problem_type if isinstance(problem_type, str) else ""
		error_code, message = problems.get(key, _UNKNOWN_QUERY_PROBLEM)
		log.warning(
		"SRM could not fulfil query",
		problem_type=problem_type,
		mapped_code=error_code,
		detail=problem.get("detail"),
		)
		raise UnprocessableEntityException(error_code=error_code, message=message)

		if response.status_code >= 400:
		log.error("SRM returned error", status=response.status_code, body=response.text)
		raise DownstreamServiceException(
		@@ -181,3 +247,38 @@ class SRMClient:
		"GET", "/internal/service-instances", params=params or None, headers=headers
		)
		return [SRMServiceInstance.model_validate(i) for i in data]

		async def retrieve_location(
		self,
		query: SRMLocationQuery,
		x_correlator: str \| None = None,
		) -> SRMLocationResult:
		headers = {"x-correlator": x_correlator} if x_correlator else None
		try:
		data = await self._request(
		"POST",
		"/internal/network-queries/location",
		json=query.model_dump(mode="json", exclude_none=True),
		headers=headers,
		problems=_QUERY_PROBLEMS,
		)
		except NotFoundException as exc:
		raise NotFoundException(
		message="Device identifier not found.",
		error_code=ErrorCode.IDENTIFIER_NOT_FOUND,
		) from exc

		try:
		return SRMLocationResult.model_validate(data)
		except ValidationError as exc:
		logger.warning(
		"SRM returned an unusable location",
		x_correlator=x_correlator,
		errors=[
		{"loc": ".".join(str(part) for part in error["loc"]), "msg": error["msg"]}
		for error in exc.errors()
		],
		)
		raise DownstreamServiceException(
		message="SRM returned an unusable location",
		) from exc

src/open_exposure_gateway/api/camara/location_retrieval/API_definitions/location-retrieval.yaml

0 → 100644

+682 −0

File added.

Preview size limit exceeded, changes collapsed.

src/open_exposure_gateway/api/camara/location_retrieval/v0_5_0/router.py

0 → 100644

+79 −0

Changes for src/open_exposure_gateway/api/camara/location_retrieval/v0_5_0/router.py: 79 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from typing import Annotated, Any

		from fastapi import APIRouter, Depends

		from open_exposure_gateway.api.camara.common import XCorrelatorHeader
		from open_exposure_gateway.api.camara.location_retrieval.v0_5_0.schemas import (
		Location,
		RetrievalLocationRequest,
		)
		from open_exposure_gateway.application.services.location_retrieval_service import (
		LocationRetrievalService,
		)
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		DownstreamServiceException,
		ErrorCode,
		ForbiddenException,
		NotFoundException,
		UnauthorizedException,
		UnprocessableEntityException,
		)
		from open_exposure_gateway.dependencies import (
		CallerContext,
		get_caller_context,
		get_location_retrieval_service,
		)
		from open_exposure_gateway.schemas.common import ErrorInfo

		# CAMARA base path: the spec serves at {apiRoot}/location-retrieval/v0.5 (wire version of
		# v0.5.0), mounted bare per ADR-0020.
		BASE_PATH = "/location-retrieval/v0.5"

		router = APIRouter(prefix=BASE_PATH)

		LocationService = Annotated[LocationRetrievalService, Depends(get_location_retrieval_service)]
		Caller = Annotated[CallerContext, Depends(get_caller_context)]

		_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,
		DownstreamServiceException,
		)
		}
		_ERROR_RESPONSES[422] = {
		"model": ErrorInfo,
		"description": UnprocessableEntityException(error_code=ErrorCode.MISSING_IDENTIFIER).message,
		}
		_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(
		"/retrieve",
		tags=["Location retrieval"],
		summary="Execute location retrieval for a user device",
		description="Retrieve the area where a certain user device is localized.",
		operation_id="retrieveLocation",
		response_model=Location,
		response_model_exclude_none=True,
		responses=_responses(400, 401, 403, 404, 422, 500, 503),
		)
		async def retrieve_location(
		request: RetrievalLocationRequest,
		service: LocationService,
		caller: Caller,
		x_correlator: XCorrelatorHeader = None,
		) -> Any:
		return await service.retrieve_location(
		request=request,
		app_provider_id=caller.app_provider_id,
		x_correlator=caller.x_correlator,
		)

src/open_exposure_gateway/api/camara/location_retrieval/v0_5_0/schemas.py

0 → 100644

+101 −0

Changes for src/open_exposure_gateway/api/camara/location_retrieval/v0_5_0/schemas.py: 101 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from enum import StrEnum
		from typing import Annotated, Literal, Optional, Union

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

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


		class AreaType(StrEnum):
		CIRCLE = "CIRCLE"
		POLYGON = "POLYGON"


		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 RetrievalLocationRequest(BaseModel):
		device: Optional[Device] = None
		# CAMARA leaves maxAge unbounded; SRM's canonical max_age_seconds is >= 0 (§E.2), so a
		# negative value is the client's 400 rather than a downstream rejection.
		maxAge: Optional[int] = Field(default=None, ge=0)
		maxSurface: Optional[int] = Field(default=None, ge=1)


		class Point(BaseModel):
		latitude: float = Field(ge=-90, le=90)
		longitude: float = Field(ge=-180, le=180)


		class Circle(BaseModel):
		areaType: Literal[AreaType.CIRCLE]
		center: Point
		radius: float = Field(ge=1, description="Distance from the center in meters")


		class Polygon(BaseModel):
		areaType: Literal[AreaType.POLYGON]
		boundary: list[Point] = Field(min_length=3, max_length=15)


		Area = Annotated[Union[Circle, Polygon], Field(discriminator="areaType")]


		class Location(BaseModel):
		lastLocationTime: AwareDatetime
		area: Area
		device: Optional[DeviceResponse] = None

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

+4 −3

Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/router.py: 4 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -72,7 +72,7 @@ async def create_qod_session(
		request=request,
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=x_correlator,
		x_correlator=caller.x_correlator,
		)


		@@ -87,11 +87,12 @@ async def create_qod_session(
		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=x_correlator,
		x_correlator=caller.x_correlator,
		)


		@@ -112,7 +113,7 @@ async def delete_qod_session(
		session_id=sessionId,
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=x_correlator,
		x_correlator=caller.x_correlator,
		)
		return Response(status_code=status.HTTP_204_NO_CONTENT)

feat: implement location retrieval api

CAMARA Location Retrieval v0.5

Merge request reports