feat: Implement Traffic Influence API for version 0.10.0 (b23e62a8) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/camara/traffic_influence/API_definitions/traffic-influence.yaml

0 → 100644

+1576 −0

File added.

Preview size limit exceeded, changes collapsed.

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

0 → 100644

+179 −0

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

		from fastapi import APIRouter, Depends, Query, Request, Response, status

		from open_exposure_gateway.api.camara.common import IdempotencyKeyHeader, XCorrelatorHeader
		from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.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/v0.10.
		BASE_PATH = "/traffic-influence/v0.10"

		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,
		http_request: Request,
		response: Response,
		x_correlator: XCorrelatorHeader = None,
		idempotency_key: IdempotencyKeyHeader = None,
		) -> Any:
		created = 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,
		)
		base = str(http_request.base_url).rstrip("/")
		response.headers["Location"] = (
		f"{base}{BASE_PATH}/traffic-influences/{created.trafficInfluenceID}"
		)
		return created


		@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/v0_10_0/schemas.py

0 → 100644

+108 −0

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 = "0.10.0"


		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