feat: add list_ready_ids_in_zone method to AppInstanceRepository (8b0c4224) · Commits · OOP / code / open-exposure-gateway

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

+10 −0

Original line number	Diff line number	Diff line
		@@ -42,6 +42,16 @@ class SqlAppInstanceRepository(AppInstanceRepository):
		row = await self._session.scalar(stmt.limit(1))
		return row is not None

		async def list_ready_ids_in_zone(
		self, app_registration_id: UUID, edge_cloud_zone_id: UUID
		) -> list[UUID]:
		stmt = select(AppInstanceRow.app_instance_id).where(
		AppInstanceRow.app_registration_id == app_registration_id,
		AppInstanceRow.edge_cloud_zone_id == edge_cloud_zone_id,
		AppInstanceRow.state == AppInstanceState.READY,
		)
		return list(await self._session.scalars(stmt))

		async def save(self, app_instance: AppInstance) -> AppInstance:
		merged = await self._session.merge(AppInstanceMapper.to_row(app_instance))
		await self._session.flush()

src/open_exposure_gateway/application/mappers/traffic_influence_mapper.py

+5 −1

Original line number	Diff line number	Diff line
		@@ -147,6 +147,7 @@ def build_traffic_influence_changed_event(
		request_metadata: dict[str, Any],
		source: str,
		occurred_at: str,
		selected_app_instance_ids: list[UUID] \| None = None,
		) -> TrafficInfluenceChangedCloudEvent:
		"""`onTrafficInfluenceChanged` -- TI's equivalent of
		quality_on_demand_mapper.build_qos_status_changed_event. `data` is the same resource
		@@ -156,9 +157,12 @@ def build_traffic_influence_changed_event(
		field nullable, so an unset optional must be omitted here too, not serialized as null.
		"""
		response = build_traffic_influence_response(traffic_influence, request_metadata)
		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]
		return TrafficInfluenceChangedCloudEvent(
		id=uuid4(),
		source=source,
		time=occurred_at,
		data=response.model_dump(mode="json", exclude_none=True),
		data=data,
		)

src/open_exposure_gateway/application/services/traffic_influence_service.py

+27 −5

Original line number	Diff line number	Diff line
		@@ -49,6 +49,7 @@ from open_exposure_gateway.ports.database.callbacks import (
		CallbackDeliveryRepository,
		CallbackRegistrationRepository,
		)
		from open_exposure_gateway.ports.database.instances import AppInstanceRepository
		from open_exposure_gateway.ports.database.operations import OperationRepository
		from open_exposure_gateway.ports.database.registration import AppRegistrationRepository
		from open_exposure_gateway.ports.database.traffic_influences import TrafficInfluenceRepository
		@@ -90,6 +91,7 @@ class TrafficInfluenceService:
		callback_delivery_repo: Optional[CallbackDeliveryRepository] = None,
		callback_delivery_port: Optional[TrafficInfluenceCallbackDeliveryPort] = None,
		app_registration_repo: Optional[AppRegistrationRepository] = None,
		app_instance_repo: Optional[AppInstanceRepository] = None,
		) -> None:
		self.srm_client = srm_client
		self._publisher = publisher
		@@ -99,6 +101,7 @@ class TrafficInfluenceService:
		self._callback_delivery_repo = callback_delivery_repo
		self._callback_delivery_port = callback_delivery_port
		self._app_registration_repo = app_registration_repo
		self._app_instance_repo = app_instance_repo

		def _parse_traffic_influence_id(self, traffic_influence_id: str) -> UUID:
		try:
		@@ -141,7 +144,10 @@ class TrafficInfluenceService:
		status=None,
		x_correlator=x_correlator,
		)
		if zones:
		if not zones:
		raise NotFoundException(
		message=f"Edge cloud region {request.edgeCloudRegion} has no zones"
		)
		return zones[0].id
		if request.appInstanceId is not None:
		instances = await self.srm_client.get_app_instances(
		@@ -150,7 +156,8 @@ class TrafficInfluenceService:
		region=None,
		x_correlator=x_correlator,
		)
		if instances:
		if not instances:
		raise NotFoundException(message=f"App instance {request.appInstanceId} not found")
		return instances[0].zone_id
		return None

		@@ -259,7 +266,7 @@ class TrafficInfluenceService:
		traffic_influence_id=traffic_influence_id,
		operation_id=operation_id,
		app_registration_id=app_registration_id,
		edge_cloud_zone_id=request.edgeCloudZoneId,
		edge_cloud_zone_id=UUID(zone_id) if zone_id is not None else None,
		edge_cloud_region=request.edgeCloudRegion,
		source_port=source_filter.sourcePort if source_filter else None,
		destination_port=destination_filter.destinationPort if destination_filter else None,
		@@ -332,11 +339,26 @@ class TrafficInfluenceService:
		f"{traffic_influence.traffic_influence_id}"
		)
		request_metadata = await self._fetch_request_metadata(registration_operation_id)
		selected_app_instance_ids: list[UUID] = []
		if (
		traffic_influence.state == RecordTrafficInfluenceState.ACTIVE
		and traffic_influence.edge_cloud_zone_id is not None
		and self._app_instance_repo is not None
		):
		selected_app_instance_ids = await self._app_instance_repo.list_ready_ids_in_zone(
		traffic_influence.app_registration_id, traffic_influence.edge_cloud_zone_id
		)
		pinned_app_instance_id = request_metadata.get("appInstanceId")
		if pinned_app_instance_id is not None:
		selected_app_instance_ids = [
		i for i in selected_app_instance_ids if i == UUID(pinned_app_instance_id)
		]
		cloud_event = build_traffic_influence_changed_event(
		traffic_influence=traffic_influence,
		request_metadata=request_metadata,
		source=source,
		occurred_at=occurred_at,
		selected_app_instance_ids=selected_app_instance_ids,
		)

		prior_attempts = await self._callback_delivery_repo.list_by_callback_registration_id(

src/open_exposure_gateway/dependencies.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -225,6 +225,7 @@ def get_traffic_influence_service(
		get_traffic_influence_callback_client
		),
		app_registration_repo: AppRegistrationRepository = Depends(get_app_registration_repo),
		app_instance_repo: AppInstanceRepository = Depends(get_app_instance_repo),
		) -> TrafficInfluenceService:
		return TrafficInfluenceService(
		srm,
		@@ -235,4 +236,5 @@ def get_traffic_influence_service(
		callback_delivery_repo,
		callback_delivery_port,
		app_registration_repo,
		app_instance_repo,
		)

src/open_exposure_gateway/main.py

+1 −0

Original line number	Diff line number	Diff line
		@@ -134,6 +134,7 @@ def _build_traffic_influence_service(
		callback_delivery_repo=SqlCallbackDeliveryRepository(session),
		callback_delivery_port=traffic_influence_callback_client,
		app_registration_repo=SqlAppRegistrationRepository(session),
		app_instance_repo=SqlAppInstanceRepository(session),
		)