feat: implement POST /traffic-influence-devices (3c2523d3) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/router.py

+25 −4

Changes for src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/router.py: 25 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -61,6 +61,7 @@ _ERROR_RESPONSES: dict[int \| str, dict[str, Any]] = {
		DownstreamServiceException,
		)
		}
		_ERROR_RESPONSES[422] = {"model": ErrorInfo, "description": "Unprocessable content"}
		_ERROR_RESPONSES[500] = {"model": ErrorInfo, "description": "Internal server error"}


		@@ -170,12 +171,32 @@ async def delete_traffic_influence(
		tags=["Traffic Influence API write"],
		summary="Creates a new TrafficInfluence resource influencing the traffic "
		"toward local instances of the Application for a specific user.",
		responses=_responses(400, 401, 403, 404, 409, 500, 501, 503),
		response_model=TrafficInfluence,
		response_model_exclude_none=True,
		status_code=status.HTTP_201_CREATED,
		responses=_responses(400, 401, 403, 404, 409, 422, 500, 503),
		)
		async def post_traffic_influence_device(request: PostTrafficInfluenceDevice) -> Any:
		raise NotImplementedException(
		message="Device-scoped traffic influence is not implemented in this release"
		async def post_traffic_influence_device(
		request: PostTrafficInfluenceDevice,
		service: TrafficInfluenceServiceDep,
		caller: Caller,
		http_request: Request,
		response: Response,
		x_correlator: XCorrelatorHeader = None,
		idempotency_key: IdempotencyKeyHeader = None,
		) -> Any:
		created = await service.create_traffic_influence_device(
		request=request,
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=x_correlator,
		idempotency_key=idempotency_key,
		)
		base = str(http_request.base_url).rstrip("/")
		response.headers["Location"] = (
		f"{base}{BASE_PATH}/traffic-influences/{created.trafficInfluenceID}"
		)
		return created


		@router.patch(

src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/schemas.py

+57 −10

Changes for src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/schemas.py: 57 added lines, 10 removed lines.
Original line number	Diff line number	Diff line
		@@ -3,7 +3,9 @@ from enum import StrEnum
		from typing import Any, Literal, Optional
		from uuid import UUID

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

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

		SOURCE_API = "traffic-influence"
		SOURCE_API_VERSION = "0.10.0"
		@@ -85,19 +87,64 @@ class PostTrafficInfluence(BaseTrafficInfluence):
		pass


		class Device(BaseModel):
		"""CAMARA `Device` -- the subscriber a device-scoped policy applies to. Modelled
		loosely: `POST /traffic-influence-devices` is a 501 stub, so nothing reads these
		fields yet. Tighten to the full DeviceIpv4Addr/PhoneNumber shapes when it is served.
		"""
		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)

		phoneNumber: Optional[str] = None
		@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[dict[str, Any]] = None
		ipv4Address: Optional[DeviceIpv4Addr] = None
		ipv6Address: Optional[str] = None


		class PostTrafficInfluenceDevice(BaseTrafficInfluence):
		@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 PostTrafficInfluenceDevice(PostTrafficInfluence):
		device: Optional[Device] = None

src/open_exposure_gateway/application/mappers/traffic_influence_mapper.py

+32 −1

Changes for src/open_exposure_gateway/application/mappers/traffic_influence_mapper.py: 32 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -6,13 +6,18 @@ from pydantic import ValidationError
		from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		SOURCE_API,
		SOURCE_API_VERSION,
		Device,
		DeviceResponse,
		PostTrafficInfluence,
		)
		from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		TrafficInfluence as TrafficInfluenceResponse,
		)
		from open_exposure_gateway.domain.models import TrafficInfluence as TrafficInfluenceRecord
		from open_exposure_gateway.domain.quality_on_demand import SourceSpec
		from open_exposure_gateway.domain.quality_on_demand import (
		NetworkCapabilityTargetDevice,
		SourceSpec,
		)
		from open_exposure_gateway.domain.traffic_influence import (
		NetworkCapabilityDeactivateTarget,
		NetworkCapabilityParameters,
		@@ -28,6 +33,17 @@ from open_exposure_gateway.domain.traffic_influence import (
		)


		def build_device_response(device: Device) -> DeviceResponse \| None:
		"""CAMARA returns exactly one identifier: phone number, then IPv4, then IPv6."""
		if device.phoneNumber is not None:
		return DeviceResponse.model_construct(phoneNumber=device.phoneNumber)
		if device.ipv4Address is not None:
		return DeviceResponse.model_construct(ipv4Address=device.ipv4Address)
		if device.ipv6Address is not None:
		return DeviceResponse.model_construct(ipv6Address=device.ipv6Address)
		return None


		def build_activate_command(
		request: PostTrafficInfluence,
		operation_id: UUID,
		@@ -37,6 +53,7 @@ def build_activate_command(
		service_instance_id: UUID,
		app_provider_id: str,
		zone_id: str \| None,
		device: Device \| None = None,
		) -> SRMNetworkCapabilityActivateCommand:
		source_filter = request.sourceTrafficFilters
		destination_filter = request.destinationTrafficFilters
		@@ -46,6 +63,16 @@ def build_activate_command(
		service_specification_ref=str(request.appId),
		external_app_id=str(request.appId),
		),
		device=NetworkCapabilityTargetDevice(
		phone_number=device.phoneNumber,
		ipv4=device.ipv4Address.publicAddress if device.ipv4Address else None,
		ipv4_port=device.ipv4Address.publicPort if device.ipv4Address else None,
		ipv4_private=device.ipv4Address.privateAddress if device.ipv4Address else None,
		ipv6=device.ipv6Address,
		network_access_id=device.networkAccessIdentifier,
		)
		if device is not None
		else None,
		)

		return SRMNetworkCapabilityActivateCommand(
		@@ -161,6 +188,10 @@ def build_traffic_influence_changed_event(
		data = response.model_dump(mode="json", exclude_none=True)
		if selected_app_instance_ids:
		data["selected_appInstanceId"] = [str(i) for i in selected_app_instance_ids]
		if request_metadata.get("device") is not None:
		device_response = build_device_response(Device.model_validate(request_metadata["device"]))
		if device_response is not None:
		data["deviceResponse"] = device_response.model_dump(mode="json", exclude_none=True)
		return TrafficInfluenceChangedCloudEvent(
		id=uuid4(),
		source=source,

src/open_exposure_gateway/application/services/traffic_influence_service.py

+37 −0

Changes for src/open_exposure_gateway/application/services/traffic_influence_service.py: 37 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -7,7 +7,9 @@ from pydantic import BaseModel, ValidationError

		from open_exposure_gateway.adapters.errors import DuplicateOperationError
		from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		Device,
		PostTrafficInfluence,
		PostTrafficInfluenceDevice,
		SubscriptionProtocol,
		)
		from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		@@ -28,8 +30,10 @@ from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		DownstreamServiceException,
		ErrorCode,
		NotFoundException,
		NotImplementedException,
		UnprocessableEntityException,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		SRMOperationCompleted,
		@@ -224,6 +228,37 @@ class TrafficInfluenceService:
		traffic_influence, operation.metadata.get("request", {})
		)

		async def create_traffic_influence_device(
		self,
		request: PostTrafficInfluenceDevice,
		tenant_id: str,
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		idempotency_key: Optional[str] = None,
		) -> TrafficInfluenceResponse:
		device = request.device
		if device is None:
		raise UnprocessableEntityException(
		error_code=ErrorCode.MISSING_IDENTIFIER,
		message="The device cannot be identified.",
		)
		if device.networkAccessIdentifier is not None:
		if not (device.phoneNumber or device.ipv4Address or device.ipv6Address):
		raise UnprocessableEntityException(
		error_code=ErrorCode.SERVICE_NOT_APPLICABLE,
		message="The service is not available for the provided identifier.",
		)
		device = device.model_copy(update={"networkAccessIdentifier": None})

		return await self.create_traffic_influence(
		request=request,
		tenant_id=tenant_id,
		app_provider_id=app_provider_id,
		x_correlator=x_correlator,
		idempotency_key=idempotency_key,
		device=device,
		)

		async def create_traffic_influence(
		self,
		request: PostTrafficInfluence,
		@@ -231,6 +266,7 @@ class TrafficInfluenceService:
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		idempotency_key: Optional[str] = None,
		device: Optional[Device] = None,
		) -> TrafficInfluenceResponse:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")
		@@ -283,6 +319,7 @@ class TrafficInfluenceService:
		service_instance_id=traffic_influence_id,
		app_provider_id=app_provider_id,
		zone_id=zone_id,
		device=device,
		)

		operation = Operation(

src/open_exposure_gateway/domain/traffic_influence.py

+5 −1

Changes for src/open_exposure_gateway/domain/traffic_influence.py: 5 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -6,7 +6,10 @@ from uuid import UUID

		from pydantic import BaseModel, Field

		from open_exposure_gateway.domain.quality_on_demand import SourceSpec
		from open_exposure_gateway.domain.quality_on_demand import (
		NetworkCapabilityTargetDevice,
		SourceSpec,
		)


		class Subject(StrEnum):
		@@ -26,6 +29,7 @@ class NetworkCapabilityTargetApplicationReference(BaseModel):

		class NetworkCapabilityTarget(BaseModel):
		application_reference: NetworkCapabilityTargetApplicationReference
		device: NetworkCapabilityTargetDevice \| None = None


		class TrafficFilter(BaseModel):