Metadata-Version: 2.4
Name: weni-eda
Version: 0.3.0a1
Summary: Python library to simplify Event-Driven Architecture (EDA) with Django and RabbitMQ
License-Expression: MPL-2.0
License-File: LICENSE
Keywords: eda,event-driven,rabbitmq,amqp,django
Author: Weni
Requires-Python: >=3.8
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: Django
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Dist: amqp (>=5.2.0,<6.0.0)
Project-URL: Repository, https://github.com/weni-ai/weni-eda
Description-Content-Type: text/markdown

# Weni EDA

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

| Broker | SSL | Params factory | Env prefix |
|--------|-----|----------------|------------|
| **RabbitMQ** | No | `ConnectionParamsFactory` | `EDA_*` |
| **AmazonMQ** | Yes (port 5671) | `AMQConnectionParamsFactory` | `AMQ_*` |

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

```sh
pip install weni-eda
```

## Django setup

Add the app to `INSTALLED_APPS`:

```py
# 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):

```py
# 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

| Variable | Example | Description |
|----------|---------|-------------|
| `EDA_BROKER_HOST` | `"localhost"` | Broker hostname or IP |
| `EDA_BROKER_PORT` | `5672` | Broker port |
| `EDA_BROKER_USER` | `"guest"` | Username |
| `EDA_BROKER_PASSWORD` | `"guest"` | Password |
| `EDA_VIRTUAL_HOST` | `"/"` | Virtual host |
| `EDA_PRODUCER` | `"weni-engine"` | Producer name included in event envelopes |
| `EDA_CONSUMERS_HANDLE` | `"myapp.messages.handle.handle_consumers"` | Consumer registration function |

### Publisher

```py
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",
)
```

To publish a standardized event envelope, pass `event_type`. The library wraps the payload with `event_id`, `producer` (from `EDA_PRODUCER`), and `timestamp`:

```py
publisher.send_message(
    {"uuid": "8e7d8a", "name": "Novo Nome do Projeto"},
    exchange="projects",
    routing_key="project.updated",
    event_type="project.updated",
)
```

`ConnectionParamsFactory` reads the `EDA_*` settings above.

### Consumer

1. Implement a consumer:

```py
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()
```

For event envelopes, use `message.event()` or `message.data()`:

```py
class ProjectUpdatedConsumer(EDAConsumer):
    def consume(self, message: Message):
        event = message.event()
        # event.event_type, event.producer, event.timestamp, event.data
        self.ack()
```

2. Register it in `handle_consumers`:

```py
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):

```sh
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

| Variable | Example | Description |
|----------|---------|-------------|
| `AMQ_BROKER_HOST` | `"b-xxxx.mq.us-east-1.amazonaws.com"` | Broker hostname |
| `AMQ_BROKER_PORT` | `5671` | SSL port (default `5671`) |
| `AMQ_BROKER_USER` | `"myuser"` | Username |
| `AMQ_BROKER_PASSWORD` | `"mypassword"` | Password |
| `AMQ_VIRTUAL_HOST` | `"/"` | Virtual host |
| `AMQ_BROKER_HEARTBEAT` | `300` | Heartbeat interval in seconds (default `300`) |
| `AMQ_BROKER_SSL_SERVER_HOSTNAME` | `"b-xxxx.mq.us-east-1.amazonaws.com"` | Hostname for SSL certificate verification / SNI (defaults to `AMQ_BROKER_HOST`) |

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

### Publisher

```py
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:

```sh
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`):

| Method / attribute | Description |
|--------------------|-------------|
| `message.body` | Raw body (`bytes`) |
| `message.json(encoding="utf-8")` | Parse body as JSON → `dict` |
| `message.event(encoding="utf-8")` | Parse body as event envelope → `Event` |
| `message.data(encoding="utf-8")` | Return `event.data` from an event envelope |
| `self.ack()` | Ack the message (remove from queue) |
| `message.reject(requeue=False)` | Reject the message (called automatically if `consume` raises) |

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

### `Channel` API

`handle_consumers` receives a `weni.eda.channels.Channel`:

| Method | Description |
|--------|-------------|
| `channel.basic_consume(queue, callback=...)` | Register a queue consumer |
| `channel.basic_qos(...)` | Set prefetch limits before consuming |

### `edaconsume` flags

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

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

| Flag | Default | Description |
|------|---------|-------------|
| `--params-class` | `ConnectionParamsFactory` (RabbitMQ) | Dotted path to the params factory |
| `--handle` | `settings.EDA_CONSUMERS_HANDLE` | Dotted path to `handle_consumers(channel)` |
| `--backend` | `settings.EDA_CONNECTION_BACKEND` or `PyAMQPConnectionBackend` | Connection backend |

Quick reference:

```sh
# 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`)

```py
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:

```sh
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](LICENSE) file for the full text.

