feat: implement location retrieval API and related schemas (de444148) · Commits · OOP / code / open-exposure-gateway

.env.example

+1 −0

Original line number	Diff line number	Diff line
		@@ -24,3 +24,4 @@

		# CALLBACK_SETTINGS__TIMEOUT=10.0
		# QOD_SETTINGS__SERVICE_SPECIFICATION_ID="7608e902-b927-559f-b448-e7e9061dfa5c"
		# LOCATION_RETRIEVAL_SETTINGS__SERVICE_SPECIFICATION_ID="0129e7ce-8e02-5dfd-bae1-0303bcc7676e"

src/open_exposure_gateway/adapters/http/srm_client.py

+44 −0

Original line number	Diff line number	Diff line
		@@ -7,7 +7,9 @@ import structlog
		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 +17,22 @@ 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__)

		_QUERY_PROBLEM_CODES: dict[str, ErrorCode] = {
		"unable_to_locate": ErrorCode.LOCATION_RETRIEVAL_UNABLE_TO_LOCATE,
		"unable_to_fulfill_max_age": ErrorCode.LOCATION_RETRIEVAL_UNABLE_TO_FULFILL_MAX_AGE,
		"unable_to_fulfill_max_surface": (ErrorCode.LOCATION_RETRIEVAL_UNABLE_TO_FULFILL_MAX_SURFACE),
		"unsupported_identifier": ErrorCode.UNSUPPORTED_IDENTIFIER,
		"service_not_applicable": ErrorCode.SERVICE_NOT_APPLICABLE,
		}


		class SRMClient:
		def __init__(self) -> None:
		@@ -55,6 +69,16 @@ class SRMClient:
		log.warning("SRM resource not found", url=url)
		raise NotFoundException(message="Resource not found")

		if response.status_code == 422:
		problem = response.json()
		code = problem.get("code")
		error_code = _QUERY_PROBLEM_CODES.get(code, ErrorCode.SERVICE_NOT_APPLICABLE)
		log.warning("SRM could not fulfil query", code=code, mapped_code=error_code)
		raise UnprocessableEntityException(
		error_code=error_code,
		message=problem.get("detail") or "Unable to fulfil the request",
		)

		if response.status_code >= 400:
		log.error("SRM returned error", status=response.status_code, body=response.text)
		raise DownstreamServiceException(
		@@ -181,3 +205,23 @@ 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",
		"/network-queries/location",
		json=query.model_dump(mode="json", exclude_none=True),
		headers=headers,
		)
		except NotFoundException as exc:
		raise NotFoundException(
		message="Device identifier not found.",
		error_code=ErrorCode.IDENTIFIER_NOT_FOUND,
		) from exc
		return SRMLocationResult.model_validate(data)

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

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=x_correlator,
		)

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

0 → 100644

+87 −0

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, model_validator


		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)

		@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

		@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
		maxAge: Optional[int] = None
		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