feat(traffic-influence): wire router, DI, and completion dispatcher (a579aa78) · Commits · OOP / code / open-exposure-gateway

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

0 → 100644

+136 −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, status

		from open_exposure_gateway.api.camara.common import XCorrelatorHeader
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		PostTrafficInfluence,
		TrafficInfluence,
		)
		from open_exposure_gateway.application.services.traffic_influence_service import (
		TrafficInfluenceService,
		)
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		DownstreamServiceException,
		ForbiddenException,
		NotFoundException,
		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,
		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],
		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,
		) -> 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,
		)


		@router.get(
		"/traffic-influences/{trafficInfluenceID}",
		tags=["Traffic Influence API read"],
		summary="Reads a specific TrafficInfluence resource identified by the "
		"trafficInfluenceID value.",
		response_model=TrafficInfluence,
		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,
		)

src/open_exposure_gateway/dependencies.py

+25 −0

Original line number	Diff line number	Diff line
		@@ -24,6 +24,9 @@ from open_exposure_gateway.adapters.database.repos.operations import (
		from open_exposure_gateway.adapters.database.repos.qod_sessions import (
		SqlQodSessionRepository,
		)
		from open_exposure_gateway.adapters.database.repos.traffic_influences import (
		SqlTrafficInfluenceRepository,
		)
		from open_exposure_gateway.api.camara.common import XCorrelatorHeader
		from open_exposure_gateway.application.services.edge_application_management_service import (
		EdgeApplicationManagementService,
		@@ -31,6 +34,9 @@ from open_exposure_gateway.application.services.edge_application_management_serv
		from open_exposure_gateway.application.services.quality_on_demand_service import (
		QualityOnDemandService,
		)
		from open_exposure_gateway.application.services.traffic_influence_service import (
		TrafficInfluenceService,
		)
		from open_exposure_gateway.core.state import AppState
		from open_exposure_gateway.ports.database.callbacks import (
		CallbackDeliveryRepository,
		@@ -40,6 +46,7 @@ from open_exposure_gateway.ports.database.instances import AppInstanceRepository
		from open_exposure_gateway.ports.database.operations import OperationRepository
		from open_exposure_gateway.ports.database.qod_sessions import QodSessionRepository
		from open_exposure_gateway.ports.database.registration import AppRegistrationRepository
		from open_exposure_gateway.ports.database.traffic_influences import TrafficInfluenceRepository
		from open_exposure_gateway.ports.databus_port import DataBusPort
		from open_exposure_gateway.ports.qod_callback_port import QodCallbackDeliveryPort
		from open_exposure_gateway.ports.srm_port import SRMClientPort
		@@ -118,6 +125,10 @@ def get_app_instance_repo(session: SessionDep) -> AppInstanceRepository:
		return SqlAppInstanceRepository(session)


		def get_traffic_influence_repo(session: SessionDep) -> TrafficInfluenceRepository:
		return SqlTrafficInfluenceRepository(session)


		def get_callback_registration_repo(session: SessionDep) -> CallbackRegistrationRepository:
		return SqlCallbackRegistrationRepository(session)

		@@ -174,3 +185,17 @@ def get_qod_service(
		callback_delivery_repo,
		callback_delivery_port,
		)


		def get_traffic_influence_service(
		srm: SRMClientPort = Depends(get_client),
		publisher: DataBusPort = Depends(get_publisher),
		operation_repo: OperationRepository = Depends(get_operation_repo),
		traffic_influence_repo: TrafficInfluenceRepository = Depends(get_traffic_influence_repo),
		callback_registration_repo: CallbackRegistrationRepository = Depends(
		get_callback_registration_repo
		),
		) -> TrafficInfluenceService:
		return TrafficInfluenceService(
		srm, publisher, operation_repo, traffic_influence_repo, callback_registration_repo
		)

src/open_exposure_gateway/main.py

+36 −4

Original line number	Diff line number	Diff line
		@@ -23,6 +23,9 @@ from open_exposure_gateway.adapters.database.repos.callback_registrations import
		)
		from open_exposure_gateway.adapters.database.repos.operations import SqlOperationRepository
		from open_exposure_gateway.adapters.database.repos.qod_sessions import SqlQodSessionRepository
		from open_exposure_gateway.adapters.database.repos.traffic_influences import (
		SqlTrafficInfluenceRepository,
		)
		from open_exposure_gateway.adapters.databus.nats_adapter import (
		NatsMessagePublisher,
		NatsOperationConsumer,
		@@ -37,6 +40,9 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.router im
		from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.router import (
		router as quality_on_demand_router,
		)
		from open_exposure_gateway.api.camara.traffic_influence.vwip.router import (
		router as traffic_influence_router,
		)
		from open_exposure_gateway.api.error_handlers import (
		register_exception_handlers,
		x_correlator_header,
		@@ -48,6 +54,9 @@ from open_exposure_gateway.application.services.edge_application_management_serv
		from open_exposure_gateway.application.services.quality_on_demand_service import (
		QualityOnDemandService,
		)
		from open_exposure_gateway.application.services.traffic_influence_service import (
		TrafficInfluenceService,
		)
		from open_exposure_gateway.core.config import get_settings
		from open_exposure_gateway.core.logging import configure_logging
		from open_exposure_gateway.domain.edge_application_management import (
		@@ -102,16 +111,30 @@ def _build_qod_service(
		)


		def _build_qod_operation_completed_handler(
		def _build_operation_completed_handler(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		qod_callback_client: QodCallbackDeliveryPort,
		) -> Callable[[SRMOperationCompleted], Awaitable[None]]:
		"""Dispatches one event.srm.operation.completed to every domain that might own it.

		Each service's handle_completed self-filters by operation_type and no-ops if the
		event isn't its own, so it's safe to call all of them from a single consumer --
		this is the one subscription for a subject shared across domains (srm/
		interface-contract.md §B.5/§D.3), not one consumer per domain.
		"""

		async def handle(event: SRMOperationCompleted) -> None:
		async with session_maker() as session:
		try:
		service = _build_qod_service(session, srm_client, qod_callback_client)
		await service.handle_completed(event)
		qod_service = _build_qod_service(session, srm_client, qod_callback_client)
		traffic_influence_service = TrafficInfluenceService(
		srm_client=srm_client,
		operation_repo=SqlOperationRepository(session),
		traffic_influence_repo=SqlTrafficInfluenceRepository(session),
		)
		await qod_service.handle_completed(event)
		await traffic_influence_service.handle_completed(event)
		await session.commit()
		except Exception:
		await session.rollback()
		@@ -151,6 +174,14 @@ openapi_tags = [
		"name": "Quality on Demand Functions",
		"description": "Quality on Demand session management",
		},
		{
		"name": "Traffic Influence API read",
		"description": "Reads existing TrafficInfluence resources",
		},
		{
		"name": "Traffic Influence API write",
		"description": "Creates or modifies a TrafficInfluence resource",
		},
		{
		"name": "Platform",
		"description": "Platform-specific endpoints (health, readiness probes)",
		@@ -211,7 +242,7 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		operation_completed_consumer = NatsOperationConsumer(
		client=publisher.client,
		subject=Subject.OPERATION_COMPLETED,
		handler=_build_qod_operation_completed_handler(
		handler=_build_operation_completed_handler(
		session_maker, srm_client, qod_callback_client
		),
		)
		@@ -272,6 +303,7 @@ def create_app(lifespan: Optional[Lifespan[FastAPI]] = None) -> FastAPI:
		app.include_router(health_router, prefix="/platform")
		app.include_router(edge_application_management_router)
		app.include_router(quality_on_demand_router)
		app.include_router(traffic_influence_router)

		return app

Admin message