fix: align TrafficInfluence GET/POST with SRM contract and CAMARA spec (009e370f) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/database/repos/traffic_influences.py

+7 −0

Original line number	Diff line number	Diff line
		@@ -25,6 +25,13 @@ class SqlTrafficInfluenceRepository(TrafficInfluenceRepository):
		row = await self._session.scalar(stmt)
		return TrafficInfluenceMapper.to_domain(row) if row is not None else None

		async def list_by_app_id(self, app_id: UUID \| None) -> list[TrafficInfluence]:
		stmt = select(TrafficInfluenceRow)
		if app_id is not None:
		stmt = stmt.where(TrafficInfluenceRow.app_id == app_id)
		rows = await self._session.scalars(stmt)
		return [TrafficInfluenceMapper.to_domain(row) for row in rows]

		async def save(self, traffic_influence: TrafficInfluence) -> TrafficInfluence:
		merged = await self._session.merge(TrafficInfluenceMapper.to_row(traffic_influence))
		await self._session.flush()

src/open_exposure_gateway/adapters/http/srm_client.py

+5 −7

Original line number	Diff line number	Diff line
		@@ -5,9 +5,6 @@ import httpx
		import structlog
		from pydantic import ValidationError

		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		TrafficInfluence,
		)
		from open_exposure_gateway.core.config import get_settings
		from open_exposure_gateway.core.exceptions import (
		DownstreamServiceException,
		@@ -26,6 +23,7 @@ from open_exposure_gateway.domain.location_retrieval import (
		SRMLocationResult,
		)
		from open_exposure_gateway.domain.quality_on_demand import SRMNetworkCapability
		from open_exposure_gateway.domain.traffic_influence import SRMTrafficInfluenceCapability

		logger = structlog.get_logger(__name__)

		@@ -209,18 +207,18 @@ class SRMClient:
		self,
		traffic_influence_id: str,
		x_correlator: str \| None = None,
		) -> TrafficInfluence:
		) -> SRMTrafficInfluenceCapability:
		headers = {"x-correlator": x_correlator} if x_correlator else None
		data = await self._request(
		"GET", f"/internal/network-capabilities/{traffic_influence_id}", headers=headers
		)
		return TrafficInfluence.model_validate(data)
		return SRMTrafficInfluenceCapability.model_validate(data)

		async def get_traffic_influences(
		self,
		app_id: UUID \| None = None,
		x_correlator: str \| None = None,
		) -> list[TrafficInfluence]:
		) -> list[SRMTrafficInfluenceCapability]:
		params = {"capability_type": "traffic_influence"}
		if app_id is not None:
		params["app_id"] = str(app_id)
		@@ -228,7 +226,7 @@ class SRMClient:
		data = await self._request(
		"GET", "/internal/network-capabilities", params=params, headers=headers
		)
		return [TrafficInfluence.model_validate(t) for t in data]
		return [SRMTrafficInfluenceCapability.model_validate(t) for t in data]

		async def get_apps(self, x_correlator: str \| None = None) -> list[SRMCatalogPayload]:
		headers = {"x-correlator": x_correlator} if x_correlator else None

src/open_exposure_gateway/application/mappers/traffic_influence_mapper.py

+25 −0

Original line number	Diff line number	Diff line
		from typing import Any
		from uuid import UUID

		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import PostTrafficInfluence
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		TrafficInfluence as TrafficInfluenceResponse,
		)
		from open_exposure_gateway.domain.models import TrafficInfluence as TrafficInfluenceRecord
		from open_exposure_gateway.domain.traffic_influence import (
		NetworkCapabilityDeactivateTarget,
		NetworkCapabilityParameters,
		@@ -76,3 +81,23 @@ def build_deactivate_command(
		grace_period_seconds=grace_period_seconds,
		),
		)


		def build_traffic_influence_response(
		traffic_influence: TrafficInfluenceRecord,
		request_metadata: dict[str, Any],
		) -> TrafficInfluenceResponse:
		"""Read-side counterpart of `build_activate_command`. SRM's capability_instance for a
		traffic-influence policy never carries apiConsumerId/appInstanceId/edgeCloudRegion/
		subscriptionRequest (they have no `srm.params/v1` field of their own), so the CAMARA
		response is reconstructed from the original request, stashed on
		`operations.metadata["request"]` at POST time, with the current `trafficInfluenceID`/
		`state` from OEG's own bookkeeping overlaid on top.
		"""
		return TrafficInfluenceResponse.model_validate(
		{
		**request_metadata,
		"trafficInfluenceID": traffic_influence.traffic_influence_id,
		"state": traffic_influence.state.value,
		}
		)

src/open_exposure_gateway/application/services/traffic_influence_service.py

+68 −16

Original line number	Diff line number	Diff line
		from datetime import datetime, timezone
		from typing import Optional
		from typing import Any, Optional
		from uuid import UUID, uuid4

		import structlog
		from pydantic import BaseModel
		from pydantic import BaseModel, ValidationError

		from open_exposure_gateway.adapters.errors import DuplicateOperationError
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		@@ -18,6 +18,7 @@ from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		from open_exposure_gateway.application.mappers.traffic_influence_mapper import (
		build_activate_command,
		build_deactivate_command,
		build_traffic_influence_response,
		)
		from open_exposure_gateway.core.exceptions import (
		ConflictException,
		@@ -109,6 +110,15 @@ class TrafficInfluenceService:
		)
		if zones:
		return zones[0].id
		if request.appInstanceId is not None:
		instances = await self.srm_client.get_app_instances(
		app_id=None,
		app_instance_id=request.appInstanceId,
		region=None,
		x_correlator=x_correlator,
		)
		if instances:
		return instances[0].zone_id
		return None

		async def _replay_create(self, operation: Operation) -> TrafficInfluenceResponse:
		@@ -125,12 +135,8 @@ class TrafficInfluenceService:
		raise DownstreamServiceException(
		message="Idempotent replay could not locate the original traffic influence"
		)
		return TrafficInfluenceResponse.model_validate(
		{
		**operation.metadata.get("request", {}),
		"trafficInfluenceID": traffic_influence.traffic_influence_id,
		"state": traffic_influence.state.value,
		}
		return build_traffic_influence_response(
		traffic_influence, operation.metadata.get("request", {})
		)

		async def create_traffic_influence(
		@@ -225,6 +231,11 @@ class TrafficInfluenceService:
		api_family=_API_FAMILY,
		sink=request.subscriptionRequest.sink,
		event_types=list(request.subscriptionRequest.types),
		sink_credential_ref=(
		f"secret://oeg/{operation_id}/notification-auth-token"
		if request.subscriptionRequest.sinkCredential
		else None
		),
		expires_at=request.subscriptionRequest.config.subscriptionExpireTime,
		)
		)
		@@ -369,25 +380,66 @@ class TrafficInfluenceService:
		# Delivery of onTrafficInfluenceChanged is a separate, not-yet-built concern
		# (same scoping as the subscriptionRequest note in create_traffic_influence).

		async def _fetch_request_metadata(self, operation_id: UUID) -> dict[str, Any]:
		if self._operation_repo is None:
		return {}
		operation = await self._operation_repo.get_by_id(operation_id)
		return operation.metadata.get("request", {}) if operation is not None else {}

		async def get_traffic_influence(
		self,
		traffic_influence_id: str,
		x_correlator: Optional[str] = None,
		) -> TrafficInfluenceResponse:
		return await self.srm_client.get_traffic_influence(
		if self._traffic_influence_repo is None:
		raise RuntimeError("TrafficInfluence repository is not available")

		traffic_influence = await self._traffic_influence_repo.get_by_id(UUID(traffic_influence_id))
		if traffic_influence is None:
		raise NotFoundException(message=f"Traffic influence {traffic_influence_id} not found")

		# Best-effort: SRM's capability_instance for this policy carries no field that
		# CAMARA's TrafficInfluence needs (see build_traffic_influence_response), and its
		# read endpoint is still unconfirmed (srm_client.py), so a failure here must never
		# block a GET that OEG's own bookkeeping can already answer -- including the
		# 'deletion in progress'/'deleted' terminal states, where SRM may have already
		# purged the capability_instance entirely.
		try:
		await self.srm_client.get_traffic_influence(
		traffic_influence_id=traffic_influence_id,
		x_correlator=x_correlator,
		)
		except (NotFoundException, DownstreamServiceException, ValidationError):
		logger.warning(
		"srm_traffic_influence_read_failed", traffic_influence_id=traffic_influence_id
		)

		request_metadata = await self._fetch_request_metadata(traffic_influence.operation_id)
		return build_traffic_influence_response(traffic_influence, request_metadata)

		async def list_traffic_influences(
		self,
		app_id: Optional[UUID] = None,
		x_correlator: Optional[str] = None,
		) -> list[TrafficInfluenceResponse]:
		return await self.srm_client.get_traffic_influences(
		if self._traffic_influence_repo is None:
		raise RuntimeError("TrafficInfluence repository is not available")

		records = await self._traffic_influence_repo.list_by_app_id(app_id)

		try:
		await self.srm_client.get_traffic_influences(
		app_id=app_id,
		x_correlator=x_correlator,
		)
		except (NotFoundException, DownstreamServiceException, ValidationError):
		logger.warning("srm_traffic_influence_list_read_failed", app_id=str(app_id))

		responses = []
		for record in records:
		request_metadata = await self._fetch_request_metadata(record.operation_id)
		responses.append(build_traffic_influence_response(record, request_metadata))
		return responses

		async def delete_traffic_influence(
		self,

src/open_exposure_gateway/ports/database/traffic_influences.py

+4 −0

Original line number	Diff line number	Diff line
		@@ -15,6 +15,10 @@ class TrafficInfluenceRepository(ABC):
		async def get_by_operation_id(self, operation_id: UUID) -> TrafficInfluence \| None:
		pass

		@abstractmethod
		async def list_by_app_id(self, app_id: UUID \| None) -> list[TrafficInfluence]:
		pass

		@abstractmethod
		async def save(self, traffic_influence: TrafficInfluence) -> TrafficInfluence:
		pass