eventsender 1.4.1

Installation and configuration

The following settings should be supplied either as environment variables or via Django settings module:

EVENT_QUEUE_URL = os.environ.get('EVENT_QUEUE_URL')
EVENT_QUEUE_EXCHANGE = os.environ.get('EVENT_QUEUE_EXCHANGE')
EVENT_QUEUE_ROUTING_KEY = os.environ.get('EVENT_QUEUE_ROUTING_KEY', '')

EVENT_QUEUE_ROUTING_KEY is optional and can be omitted (it will be assumed blank by default).

Usage

Example:

from eventsender import send_event
send_event({'type': 'user.subscribed', 'username': 'john_doe'})

Or

EVENT_QUEUE_EXCHANGE=some_exchange EVENT_QUEUE_URL=amqp://guest:guest@localhost:5672/host python -c \
     "import eventsender; eventsender.send_event({'type': 'event_type', 'somedata': {'key': 'value'}})"

To publish a batch, use send_events. It sends the same messages as calling send_event for each of them, but over a single connection instead of one per event:

from eventsender import send_events
send_events([
    {'type': 'user.subscribed', 'username': 'john_doe'},
    {'type': 'user.subscribed', 'username': 'jane_doe'},
])

Tracing

send_event and send_events create an OpenTelemetry producer span per event and inject the current trace context into the AMQP message headers, so consumers can continue the trace. This only depends on opentelemetry-api, for applications without a configured OpenTelemetry backend all tracing calls are no-ops and the behaviour is unchanged. To get the spans into your tracing backend, configure an OpenTelemetry SDK in your application (for example by enabling the Sentry SDK’s OpenTelemetry support).

Running tests

Install uv and run uv run pytest tests.