refactor: CallbackDeliveryPort.deliver to accept a CloudEvent array (dc40c832) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/http/callback_client.py

+14 −3

Original line number	Diff line number	Diff line
		import httpx
		import structlog
		from pydantic import TypeAdapter

		from open_exposure_gateway.core.config import get_settings
		from open_exposure_gateway.domain.edge_application_management import (
		AppInstanceStatusChangeCloudEvent,
		AppDeploymentStatusChangeCloudEvent,
		)
		from open_exposure_gateway.ports.callback_delivery_port import CallbackEvent

		logger = structlog.get_logger(__name__)

		_DEPLOYMENT_EVENTS_ADAPTER = TypeAdapter(list[AppDeploymentStatusChangeCloudEvent])


		class HttpCallbackClient:
		def __init__(self) -> None:
		settings = get_settings()
		self.timeout = settings.callback_settings.timeout

		async def deliver(self, sink: str, event: AppInstanceStatusChangeCloudEvent) -> None:
		async def deliver(self, sink: str, event: CallbackEvent) -> None:
		# onAppInstanceStatusChange -> one CloudEvent object;
		# onAppDeploymentStatusChange -> a JSON array of CloudEvents (ADR-0008).
		content: bytes = (
		_DEPLOYMENT_EVENTS_ADAPTER.dump_json(event)
		if isinstance(event, list)
		else event.model_dump_json().encode()
		)
		async with httpx.AsyncClient(timeout=self.timeout) as client:
		response = await client.post(
		sink,
		content=event.model_dump_json(),
		content=content,
		headers={"Content-Type": "application/cloudevents+json"},
		)
		response.raise_for_status()

+11 −2

Original line number	Diff line number	Diff line
		from typing import Protocol
		from typing import Protocol, Union

		from open_exposure_gateway.domain.edge_application_management import (
		AppDeploymentStatusChangeCloudEvent,
		AppInstanceStatusChangeCloudEvent,
		)

		# A single delivery is either one instance CloudEvent (onAppInstanceStatusChange)
		# or an array of deployment CloudEvents (onAppDeploymentStatusChange) -- both go
		# to the same sink as `application/cloudevents+json` (ADR-0008).
		CallbackEvent = Union[
		AppInstanceStatusChangeCloudEvent,
		list[AppDeploymentStatusChangeCloudEvent],
		]


		class CallbackDeliveryPort(Protocol):
		async def deliver(self, sink: str, event: AppInstanceStatusChangeCloudEvent) -> None: ...
		async def deliver(self, sink: str, event: CallbackEvent) -> None: ...

+6 −4

Original line number	Diff line number	Diff line
		@@ -38,7 +38,6 @@ from open_exposure_gateway.application.services.quality_on_demand_service import
		)
		from open_exposure_gateway.core.exceptions import ErrorCode, NotFoundException
		from open_exposure_gateway.domain.edge_application_management import (
		AppInstanceStatusChangeCloudEvent,
		SRMCatalogPayload,
		SRMCatalogServiceSpecificationCreated,
		SRMServiceInstance,
		@@ -67,7 +66,10 @@ from open_exposure_gateway.domain.quality_on_demand import (
		SRMNetworkCapability,
		)
		from open_exposure_gateway.domain.quality_on_demand import Subject as QodSubject
		from open_exposure_gateway.ports.callback_delivery_port import CallbackDeliveryPort
		from open_exposure_gateway.ports.callback_delivery_port import (
		CallbackDeliveryPort,
		CallbackEvent,
		)
		from open_exposure_gateway.ports.database.callbacks import (
		CallbackDeliveryRepository,
		CallbackRegistrationRepository,
		@@ -585,9 +587,9 @@ class FakeCallbackDeliveryRepository(CallbackDeliveryRepository):

		class FakeCallbackDeliveryPort:
		def __init__(self) -> None:
		self.delivered: list[tuple[str, AppInstanceStatusChangeCloudEvent]] = []
		self.delivered: list[tuple[str, CallbackEvent]] = []

		async def deliver(self, sink: str, event: AppInstanceStatusChangeCloudEvent) -> None:
		async def deliver(self, sink: str, event: CallbackEvent) -> None:
		self.delivered.append((sink, event))