fix: further 0.10.0 work (7b8fbbd1) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/camara/traffic_influence/vwip/API_definitions/traffic_influence.yaml

deleted100644 → 0

+0 −1576

File deleted.

Preview size limit exceeded, changes collapsed.

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

deleted100644 → 0

+0 −172

Original line number	Diff line number	Diff line
		from typing import Annotated, Any, Optional
		from uuid import UUID

		from fastapi import APIRouter, Depends, Query, status

		from open_exposure_gateway.api.camara.common import IdempotencyKeyHeader, XCorrelatorHeader
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		PatchTrafficInfluence,
		PostTrafficInfluence,
		PostTrafficInfluenceDevice,
		TrafficInfluence,
		)
		from open_exposure_gateway.application.services.traffic_influence_service import (
		TrafficInfluenceService,
		)
		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_traffic_influence_service,
		)
		from open_exposure_gateway.schemas.common import ErrorInfo

		# CAMARA base path: the spec serves at {apiRoot}/traffic-influence/vwip.
		BASE_PATH = "/traffic-influence/vwip"

		router = APIRouter(prefix=BASE_PATH)

		TrafficInfluenceServiceDep = Annotated[
		TrafficInfluenceService, Depends(get_traffic_influence_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[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.get(
		"/traffic-influences",
		tags=["Traffic Influence API read"],
		summary="Retrieves existing TrafficInfluence Resources",
		response_model=list[TrafficInfluence],
		# The CAMARA schema marks no field nullable, so an unset optional must be omitted
		# rather than serialized as null -- same rule the POST response already follows.
		response_model_exclude_none=True,
		responses=_responses(400, 401, 403, 500, 503),
		)
		async def get_all_traffic_influences(
		service: TrafficInfluenceServiceDep,
		appId: Annotated[Optional[UUID], Query()] = None,
		x_correlator: XCorrelatorHeader = None,
		) -> Any:
		return await service.list_traffic_influences(app_id=appId, x_correlator=x_correlator)


		@router.post(
		"/traffic-influences",
		tags=["Traffic Influence API write"],
		summary="Creates a new TrafficInfluence resource influencing the traffic "
		"toward local instances of the Application for any user.",
		response_model=TrafficInfluence,
		response_model_exclude_none=True,
		status_code=status.HTTP_201_CREATED,
		responses=_responses(400, 401, 403, 409, 500, 503),
		)
		async def post_traffic_influence(
		request: PostTrafficInfluence,
		service: TrafficInfluenceServiceDep,
		caller: Caller,
		x_correlator: XCorrelatorHeader = None,
		idempotency_key: IdempotencyKeyHeader = None,
		) -> Any:
		return await service.create_traffic_influence(
		request=request,
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=x_correlator,
		idempotency_key=idempotency_key,
		)


		@router.get(
		"/traffic-influences/{trafficInfluenceID}",
		tags=["Traffic Influence API read"],
		summary="Reads a specific TrafficInfluence resource identified by the "
		"trafficInfluenceID value.",
		response_model=TrafficInfluence,
		response_model_exclude_none=True,
		responses=_responses(400, 401, 403, 404, 500, 503),
		)
		async def get_traffic_influence_by_id(
		trafficInfluenceID: str,
		service: TrafficInfluenceServiceDep,
		x_correlator: XCorrelatorHeader = None,
		) -> Any:
		return await service.get_traffic_influence(
		traffic_influence_id=trafficInfluenceID,
		x_correlator=x_correlator,
		)


		@router.delete(
		"/traffic-influences/{trafficInfluenceID}",
		tags=["Traffic Influence API write"],
		summary="Delete an existing TrafficInfluence resource",
		status_code=status.HTTP_202_ACCEPTED,
		responses=_responses(400, 401, 403, 404, 409, 500, 503),
		)
		async def delete_traffic_influence(
		trafficInfluenceID: str,
		service: TrafficInfluenceServiceDep,
		caller: Caller,
		x_correlator: XCorrelatorHeader = None,
		) -> None:
		await service.delete_traffic_influence(
		traffic_influence_id=trafficInfluenceID,
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=x_correlator,
		)


		@router.post(
		"/traffic-influence-devices",
		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),
		)
		async def post_traffic_influence_device(request: PostTrafficInfluenceDevice) -> Any:
		raise NotImplementedException(
		message="Device-scoped traffic influence is not implemented in this release"
		)


		@router.patch(
		"/traffic-influences/{trafficInfluenceID}",
		tags=["Traffic Influence API write"],
		summary="Updates a specific TrafficInfluence resource, identified by the "
		"trafficInfluenceID value.",
		responses=_responses(400, 401, 403, 404, 409, 500, 501, 503),
		)
		async def patch_traffic_influence(trafficInfluenceID: str, request: PatchTrafficInfluence) -> Any:
		raise NotImplementedException(
		message="Traffic influence modification is not implemented in this release"
		)

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

deleted100644 → 0

+0 −108

Original line number	Diff line number	Diff line
		from datetime import datetime
		from enum import StrEnum
		from typing import Any, Literal, Optional
		from uuid import UUID

		from pydantic import BaseModel, Field

		SOURCE_API = "traffic-influence"
		SOURCE_API_VERSION = "wip"


		class TrafficInfluenceState(StrEnum):
		ORDERED = "ordered"
		CREATED = "created"
		ACTIVE = "active"
		ERROR = "error"
		DELETION_IN_PROGRESS = "deletion in progress"
		DELETED = "deleted"


		class SubscriptionProtocol(StrEnum):
		HTTP = "HTTP"
		MQTT3 = "MQTT3"
		MQTT5 = "MQTT5"
		AMQP = "AMQP"
		NATS = "NATS"
		KAFKA = "KAFKA"


		SubscriptionEventType = Literal["org.camaraproject.traffic-influence.v1.traffic-influence-change"]


		class Config(BaseModel):
		subscriptionDetail: dict[str, Any]
		subscriptionExpireTime: Optional[datetime] = None
		subscriptionMaxEvents: Optional[int] = Field(default=None, ge=1)
		initialEvent: Optional[bool] = None


		class SubscriptionRequest(BaseModel):
		protocol: SubscriptionProtocol
		sink: str = Field(pattern=r"^https:\/\/.+$")
		# sinkCredential is a discriminated union (PLAIN/ACCESSTOKEN/REFRESHTOKEN) in the
		# CAMARA spec; kept generic here since OEG only persists a sink_credential_ref
		# (never the raw credential) and doesn't branch on credential type.
		sinkCredential: Optional[dict[str, Any]] = None
		types: list[SubscriptionEventType] = Field(min_length=1, max_length=1)
		config: Config


		class SourceTrafficFilters(BaseModel):
		sourcePort: Optional[int] = Field(default=None, ge=0, le=65535)


		class DestinationTrafficFilters(BaseModel):
		destinationPort: Optional[int] = Field(default=None, ge=0, le=65535)
		destinationProtocol: Optional[str] = None


		class BaseTrafficInfluence(BaseModel):
		apiConsumerId: str
		appId: UUID
		appInstanceId: Optional[UUID] = None
		edgeCloudRegion: Optional[str] = None
		edgeCloudZoneId: Optional[UUID] = None
		sourceTrafficFilters: Optional[SourceTrafficFilters] = None
		destinationTrafficFilters: Optional[DestinationTrafficFilters] = None
		subscriptionRequest: Optional[SubscriptionRequest] = None


		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.
		"""

		phoneNumber: Optional[str] = None
		networkAccessIdentifier: Optional[str] = None
		ipv4Address: Optional[dict[str, Any]] = None
		ipv6Address: Optional[str] = None


		class PostTrafficInfluenceDevice(BaseTrafficInfluence):
		device: Optional[Device] = None


		class PatchTrafficInfluence(BaseModel):
		"""CAMARA marks `trafficInfluenceID`/`apiConsumerId`/`appId`/`state` readOnly on
		PATCH, so they are absent here. Note the spec still inherits
		`required: [apiConsumerId, appId]` from the base schema while marking both readOnly --
		a contradiction left permissive here rather than resolved, since PATCH is a 501 stub.
		"""

		appInstanceId: Optional[UUID] = None
		edgeCloudRegion: Optional[str] = None
		edgeCloudZoneId: Optional[UUID] = None
		sourceTrafficFilters: Optional[SourceTrafficFilters] = None
		destinationTrafficFilters: Optional[DestinationTrafficFilters] = None
		subscriptionRequest: Optional[SubscriptionRequest] = None


		class TrafficInfluence(BaseTrafficInfluence):
		trafficInfluenceID: UUID
		state: TrafficInfluenceState

src/open_exposure_gateway/application/mappers/traffic_influence_mapper.py

+5 −3

Original line number	Diff line number	Diff line
		@@ -3,12 +3,12 @@ from uuid import UUID, uuid4

		from pydantic import ValidationError

		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		SOURCE_API,
		SOURCE_API_VERSION,
		PostTrafficInfluence,
		)
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		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
		@@ -152,11 +152,13 @@ def build_traffic_influence_changed_event(
		quality_on_demand_mapper.build_qos_status_changed_event. `data` is the same resource
		representation `build_traffic_influence_response` returns (the notification schema is
		`allOf: [PostTrafficInfluenceDevice]`, i.e. the whole resource, not a delta like QoD's).
		`exclude_none` mirrors the routers' `response_model_exclude_none`: the schema marks no
		field nullable, so an unset optional must be omitted here too, not serialized as null.
		"""
		response = build_traffic_influence_response(traffic_influence, request_metadata)
		return TrafficInfluenceChangedCloudEvent(
		id=uuid4(),
		source=source,
		time=occurred_at,
		data=response.model_dump(mode="json"),
		data=response.model_dump(mode="json", exclude_none=True),
		)

src/open_exposure_gateway/application/services/traffic_influence_service.py

+17 −4

Original line number	Diff line number	Diff line
		@@ -6,13 +6,14 @@ import structlog
		from pydantic import BaseModel, ValidationError

		from open_exposure_gateway.adapters.errors import DuplicateOperationError
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		PostTrafficInfluence,
		SubscriptionProtocol,
		)
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		TrafficInfluence as TrafficInfluenceResponse,
		)
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		TrafficInfluenceState as ApiTrafficInfluenceState,
		)
		from open_exposure_gateway.application.mappers.traffic_influence_mapper import (
		@@ -190,6 +191,18 @@ class TrafficInfluenceService:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		# SubscriptionProtocol keeps CAMARA's full enum so the OpenAPI stays faithful, but
		# "Only HTTP is allowed for now" (SubscriptionProtocol description) and OEG only
		# has an HTTP delivery adapter -- accepting another value would silently deliver
		# over the wrong transport.
		if (
		request.subscriptionRequest is not None
		and request.subscriptionRequest.protocol is not SubscriptionProtocol.HTTP
		):
		raise BadRequestException(
		message="Only HTTP notification delivery is supported for subscriptionRequest"
		)

		if idempotency_key is not None:
		existing = await self._operation_repo.get_by_idempotency_key(tenant_id, idempotency_key)
		if existing is not None:
		@@ -315,7 +328,7 @@ class TrafficInfluenceService:
		raise RuntimeError("CallbackDeliveryPort/CallbackDeliveryRepository is not available")

		source = (
		f"{get_settings().public_base_url}/traffic-influence/vwip/traffic-influences/"
		f"{get_settings().public_base_url}/traffic-influence/v0.10/traffic-influences/"
		f"{traffic_influence.traffic_influence_id}"
		)
		request_metadata = await self._fetch_request_metadata(registration_operation_id)