Weni EDA

weni-eda is a Python library that simplifies Event-Driven Architecture (EDA) with Django and AMQP brokers. It supports:

Both scopes share the same consumer and publisher APIs — you only swap the connection params factory (and the matching env vars).

Features

• Easy integration with Django
• RabbitMQ (plain AMQP) and AmazonMQ (AMQP over SSL)
• Transport-agnostic Message and Channel abstractions
• Optional buffered consumers for high-throughput workloads

Installation

pip install weni-eda

Django setup

Add the app to INSTALLED_APPS:

# settings.py
INSTALLED_APPS = [
    # ...
    "weni.eda.django.eda_app",
]

Point EDA_CONSUMERS_HANDLE to the function that registers your consumers (used by both brokers unless overridden per process):

# settings.py
EDA_CONSUMERS_HANDLE = "myapp.messages.handle.handle_consumers"

RabbitMQ (no SSL)

Use this for a standard RabbitMQ broker on port 5672.

Environment variables

Publisher

from weni.eda.django import ConnectionParamsFactory
from weni.eda.eda_publisher import EDAPublisher

publisher = EDAPublisher(ConnectionParamsFactory)
publisher.send_message(
    {"event": "order.created", "order_id": 123},
    exchange="orders",
    routing_key="order.created",
)

ConnectionParamsFactory reads the EDA_* settings above.

Consumer

1. Implement a consumer:

from weni.eda.django.consumers import EDAConsumer
from weni.eda.messages import Message


class ExampleConsumer(EDAConsumer):
    def consume(self, message: Message):
        body = message.json()
        # ... handle body ...
        self.ack()

2. Register it in handle_consumers:

from weni.eda.channels import Channel
from .example_consumer import ExampleConsumer


def handle_consumers(channel: Channel):
    channel.basic_consume("example-queue", callback=ExampleConsumer().handle)

3. Start consuming (default params factory = RabbitMQ / no SSL):

python manage.py edaconsume

AmazonMQ (SSL)

Use this for AmazonMQ (or any AMQP broker that requires TLS). Connections use SSL on port 5671 via AMQConnectionParamsFactory.

Environment variables

You still need EDA_CONSUMERS_HANDLE (or --handle) so the process knows which consumers to register.

Publisher

from weni.eda.django import AMQConnectionParamsFactory
from weni.eda.eda_publisher import EDAPublisher

publisher = EDAPublisher(AMQConnectionParamsFactory)
publisher.send_message(
    {"event": "order.created", "order_id": 123},
    exchange="orders",
    routing_key="order.created",
)

AMQConnectionParamsFactory reads the AMQ_* settings and enables SSL automatically.

Consumer

Consumers and handle_consumers are identical to RabbitMQ. The only difference is which params factory you pass when starting the process:

python manage.py edaconsume \
  --params-class "weni.eda.django.AMQConnectionParamsFactory"

Consumers reference

Message API

Consumers receive a weni.eda.messages.Message (transport-agnostic — no need to import amqp):

If consume raises, the message is rejected and the error is logged.

Channel API

handle_consumers receives a weni.eda.channels.Channel:

edaconsume flags

Useful when one project talks to both brokers (or multiple consumer groups):

python manage.py edaconsume \
  --handle "myapp.messages.handle.handle_consumers" \
  --backend "weni.eda.backends.pyamqp_flush_backend.PyAMQPFlushConnectionBackend" \
  --params-class "weni.eda.django.AMQConnectionParamsFactory"

Quick reference:

# RabbitMQ (no SSL)
python manage.py edaconsume

# AmazonMQ (SSL)
python manage.py edaconsume --params-class "weni.eda.django.AMQConnectionParamsFactory"

Buffered consumers (optional)

By default each consumer acks messages one by one. For high-throughput workloads that batch DB writes, use PyAMQPFlushConnectionBackend.

Your handle_consumers must register consumers and return an iterable of flushable objects, each exposing:

• flush() — persist and ack buffered work
• flush_interval (optional float) — max seconds between flushes (default 1.0)

from weni.eda.backends.pyamqp_flush_backend import PyAMQPFlushConnectionBackend
from weni.eda.channels import Channel
from weni.eda.django import AMQConnectionParamsFactory


def handle_consumers(channel: Channel):
    consumer = BufferedConsumer()  # exposes flush() and flush_interval
    consumer.setup(channel)        # channel.basic_qos(...) + channel.basic_consume(...)
    return [consumer]


def run():
    params = AMQConnectionParamsFactory.get_params()
    PyAMQPFlushConnectionBackend(handle_consumers).start_consuming(params)

Or via the management command:

python manage.py edaconsume \
  --backend "weni.eda.backends.pyamqp_flush_backend.PyAMQPFlushConnectionBackend" \
  --params-class "weni.eda.django.AMQConnectionParamsFactory"

Returning None (or an empty iterable) disables periodic flushing. You can also set settings.EDA_CONNECTION_BACKEND.

Both backends use Python logging for connection lifecycle and errors — configure handlers in your Django app (or Sentry) as needed.

License

This project is licensed under the Mozilla Public License 2.0. See the LICENSE file for the full text.