python_outbox_core.formatters.cloudevents¶

CloudEvents 1.0 formatter for protocol-level standardization.

Per PRD specifications

Classes¶

CloudEventsFormatter

Base CloudEvents 1.0 formatter.

Module Contents¶

class python_outbox_core.formatters.cloudevents.CloudEventsFormatter(source: str, data_content_type: str = 'application/json')¶

Bases: python_outbox_core.formatters.base.IEventFormatter

Base CloudEvents 1.0 formatter.

Implements CloudEvents specification for protocol-level standardization. Can be extended for specific gateways (Kong, etc.).

Spec: https://github.com/cloudevents/spec/blob/v1.0/spec.md

Example:

>>> formatter = CloudEventsFormatter(
...     source="my-service",
...     data_content_type="application/json"
... )
>>> formatted = formatter.format(event)

source¶

data_content_type = 'application/json'¶

format(event: python_outbox_core.events.IOutboxEvent) → Dict[str, Any]¶

Format event as CloudEvents 1.0.

Args:: event: Outbox event to format
Returns:: CloudEvents 1.0 formatted dictionary

enrich_metadata(event_data: Dict[str, Any]) → Dict[str, Any]¶

Hook for subclasses to enrich metadata.

Override this in subclasses (e.g., Kong formatter) to add gateway-specific metadata.

Args:

event_data: CloudEvents formatted data

Returns:

Enriched event data

Example:

>>> class KongFormatter(CloudEventsFormatter):
...     def enrich_metadata(self, event_data):
...         event_data["kong_namespace"] = "production"
...         return event_data

get_content_type() → str¶

CloudEvents content type.

Returns:: CloudEvents JSON content type