Metadata-Version: 2.4
Name: vaitty_eventbus
Version: 0.2.7
Summary: A minimal, internal Python library for decoupled event communication using channels and wildcard-capable handlers.
Author-email: Tech Data AI <tech-data-ai@rapihogar.com>
License: MIT
Project-URL: Homepage, https://bitbucket.org/rapihogar/event-bus
Classifier: License :: OSI Approved :: MIT License
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: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Operating System :: OS Independent
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: http
Requires-Dist: requests>=2.31; extra == "http"
Provides-Extra: pubsub
Requires-Dist: google-cloud-pubsub>=2.18; extra == "pubsub"
Provides-Extra: celery
Requires-Dist: celery>=5; extra == "celery"
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.100; extra == "fastapi"
Provides-Extra: test
Requires-Dist: pytest>=8; extra == "test"
Requires-Dist: pytest-cov>=5; extra == "test"
Requires-Dist: pytest-asyncio>=0.23; extra == "test"
Provides-Extra: dev
Requires-Dist: ruff<0.17,>=0.16; extra == "dev"
Requires-Dist: pre-commit>=3.7; extra == "dev"
Provides-Extra: publish
Requires-Dist: build>=1.2; extra == "publish"
Requires-Dist: twine>=5; extra == "publish"
Dynamic: license-file

# vaitty_eventbus

The platform event-bus SDK for Vaitty/Rapihogar: a provider-neutral Python library for
decoupled, event-driven communication. Keeps a tiny `emit` / `on` / `trigger` API across a
legacy Django 3.8 monolith and modern FastAPI 3.12 services, with **GCP Pub/Sub** as the
default transport in deployed environments.

---

## 🚀 Features

- 🎯 Minimal API: `emit` (produce), `on` (subscribe), `trigger` (dispatch); `on_raw` to consume the raw envelope without parsing.
- 🧱 **Typed event DTOs** — one `@dataclass` per topic (`1 type = 1 topic`), auto-registered.
- 🔌 Pluggable channels: `console`, `file` (base, stdlib-only), `http` (via the `[http]` extra or your own `session`) and **`pubsub`** (GCP, via the `[pubsub]` extra).
- 🧵 Runtime listeners for consumers: **`SyncListener`** (Celery/Django) and **`AsyncListener`** (FastAPI/asyncio).
- ✉️ Portable 3-layer envelope (payload · library metadata · channel envelope); metadata (`event_id`, `source`, `correlation_id`, …) travels as Pub/Sub attributes.
- 🧩 Provider-neutral: the public API never exposes `google.cloud.pubsub` types. Switching bus = changing the value of `EVENTBUS_CHANNEL`.
- 🪶 One package, no forks: sync and async coexist via lazy imports + optional extras. The base install has **no mandatory dependency** (not even `requests`) and runs on Python 3.8.

---

## 📦 Installation

Install only what the service needs — the extras are additive:

```bash
# Base — console/file channels + typed events. No mandatory deps (not even
# requests). Python 3.8+. The http channel also works on base if you pass your
# own `session` (e.g. the legacy monolito, que ya trae su requests, posteando al
# bridge — así no queda forzado a bumpear requests; ver "Customizing the HTTP
# transport").
pip install vaitty-eventbus

# + HTTP channel's default transport (requests) — solo si NO pasás tu propio
#   session. Trae requests>=2.31 (piso por CVE-2023-32681).
pip install 'vaitty-eventbus[http]'

# + GCP Pub/Sub backend — producers and consumers on Pub/Sub. Python 3.9+.
pip install 'vaitty-eventbus[pubsub]'

# + FastAPI — to run the AsyncListener inside a FastAPI lifespan. Python 3.10+.
pip install 'vaitty-eventbus[fastapi]'

# Async service / bridge on Pub/Sub + FastAPI (the usual combo for a modern consumer).
pip install 'vaitty-eventbus[pubsub,fastapi]'

# + Celery — context for the SyncListener in a worker.
pip install 'vaitty-eventbus[celery]'
```

Pin it in `requirements.txt`:

```
vaitty-eventbus[pubsub,fastapi]>=0.2.0
```

| Extra | Pulls in | For | Python |
|---|---|---|---|
| *(base)* | — (stdlib only) | console/file channels, typed DTOs, `http` channel with your own `session` | 3.8+ |
| `[http]` | `requests>=2.31` | `http` channel's default transport + egress relay | 3.8+ |
| `[pubsub]` | `google-cloud-pubsub` | `pubsub` channel — emit/consume on GCP | 3.9+ |
| `[fastapi]` | `fastapi` | `AsyncListener` in a FastAPI app | 3.10+ |
| `[celery]` | `celery` | `SyncListener` in a Celery worker | 3.9+ |

> The `pubsub` channel and the async listener are imported **lazily**: a sync service never
> parses `aio.py`, and importing the library without the `[pubsub]` extra never fails — the
> factory raises a clear `ConfigError` only if you actually select the `pubsub` channel without it.
> `requests` is likewise lazy: `import vaitty_eventbus` and the console/file/pubsub channels
> never import it. The `http` channel imports it only when you don't pass your own `session`
> (so a service with its own `requests` isn't forced to bump it; install `[http]` for the default).

---

## 🛠 Configuration

Configure through environment variables (or a dict passed to `EventHandlerFactory.from_config`).
The full surface is deliberately tiny:

| Variable | Scope | What | Default |
|---|---|---|---|
| `EVENTBUS_CHANNEL` | all | `console` · `file` · `http` · `pubsub` · custom | `console` |
| `EVENTBUS_SOURCE` | all | identity of the producing service (bare service name, e.g. `connect-service`) — goes into `source` and derives subscription names | `""` |
| `GOOGLE_CLOUD_PROJECT` | `pubsub` | GCP project of the environment (used to build topic/subscription paths) | *(required for pubsub)* |
| `EVENTBUS_FILE` | `file` | log file path | `events.log` |
| `EVENTBUS_HTTP_ENDPOINT` | `http` | target URL | `http://example.com/events` |
| `EVENTBUS_HTTP_METHOD` | `http` | HTTP method | `POST` |
| `EVENTBUS_HTTP_TIMEOUT` | `http` | per-request timeout (s) | *(none)* |

There is **no** provider-named key (`EVENTBUS_PUBSUB_*` is forbidden) and **no** secret: on GKE,
credentials come from **Workload Identity**. `GOOGLE_CLOUD_PROJECT` is a standard Google var and
must be set per environment (GKE does not inject it) — without it, Pub/Sub publishes fail with
`project=None`.

```python
from vaitty_eventbus import EventHandlerFactory

handler = EventHandlerFactory.from_env()  # reads EVENTBUS_* from os.environ
# or
handler = EventHandlerFactory.from_config(
    {"EVENTBUS_CHANNEL": "pubsub", "EVENTBUS_SOURCE": "connect-service"}
)
```

---

## 🧱 Typed events (DTOs)

A typed event is a `@dataclass` subclass of `Event` that declares its `topic` as a **class
attribute** (no annotation, so dataclass does not treat it as a field). Defining it
auto-registers the bijective `topic ↔ class` mapping (`1 type = 1 topic`).

**Platform DTOs ship in the library**, under `vaitty_eventbus.catalog` — producer and consumer
import the *same* class, so the payload shape can't drift between the two ends of the bus (flat
payload, ADR-047). Importing the catalog auto-registers its topics.

```python
from vaitty_eventbus.catalog import CorePolicyDisenrolled, CorePolicyEnrolled

# Flat payload (no wrapper). Dates are ISO-8601 **with timezone**, carried as
# strings — the SDK never parses them, it only preserves the contract.
evt = CorePolicyEnrolled(
    policy_number="900001",
    type="property",
    product_id=137,
    city_id=100,
    address="Av. Siempre Viva",
    issue_date="2026-08-03T10:30:00-03:00",
    valid_from="2026-09-02T00:00:00-03:00",
    valid_until="2027-09-02T23:59:59-03:00",
)
```

To define your own, subclass `Event` and declare `topic` as a class attribute:

```python
import dataclasses
from vaitty_eventbus import Event


@dataclasses.dataclass
class VisitConfirmed(Event):
    topic = "core.visit.confirmed"  # class attribute → auto-registered
    visit_id: int
    confirmed_at: str  # ISO-8601 with tz
```

The raw form `Event("core.policy.enrolled", {...})` stays valid as an escape hatch. DTOs live
in the **base** package (pure dataclasses, no GCP dep), so a legacy 3.8 service can install the
base library just to deserialize what it receives.

---

## 🌿 Producing: `emit`

`emit` accepts a typed DTO **or** the raw `(topic, payload)` form. It builds the library
metadata (envelope) and sends through the configured channel.

```python
handler = EventHandlerFactory.from_env()  # EVENTBUS_CHANNEL=pubsub

# Typed (preferred) — `evt` is a catalog DTO (see "Typed events"):
handler.emit(evt)

# Raw escape hatch:
handler.emit("core.policy.enrolled", {"policy_number": "900001", "type": "property"})

# Optional envelope metadata as kwargs:
handler.emit(evt, correlation_id="corr-xyz", tenant_id="t1")
```

On the `pubsub` channel this publishes to the topic named after the event type
(`core.policy.enrolled`); `data` + `occurred_at` go to the message body, the rest of the
metadata to attributes. `emit` waits for the publish to be confirmed (at-least-once on the
producer side) and returns the `EventMessage`.

---

## 🎧 Consuming: `on` + `trigger`

`on(pattern, callback)` registers a handler for a topic or wildcard (`core.policy.*`, `*`).
`trigger(incoming)` resolves the typed DTO from the registry and calls each matching callback
**with the typed event instance** (not the raw payload):

```python
def on_enrolled(event: CorePolicyEnrolled):
    print(event.policy_number, event.type)  # typed fields


handler.on("core.policy.enrolled", on_enrolled)

# `incoming` may be an Event, an EventMessage, or a dict envelope:
handler.trigger(
    {"type": "core.policy.enrolled", "data": {"policy_number": "900001", "type": "property"}}
)
```

`get_matching_callbacks(event)` is available if you want to drive dispatch yourself (each
callback is yielded once even if several patterns match).

---

## 🧵 Runtime listeners (consumers on Pub/Sub)

For services that pull from Pub/Sub, the SDK ships two listeners. The set of subscriptions is
**derived from the `on()` registrations** — you never write subscription names.

**Async (FastAPI / asyncio)** — `[pubsub,fastapi]`:

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from vaitty_eventbus import EventHandlerFactory
from vaitty_eventbus.runtime import AsyncListener


@asynccontextmanager
async def lifespan(app):
    handler = EventHandlerFactory.from_env()  # EVENTBUS_CHANNEL=pubsub
    handler.on("core.policy.enrolled", on_enrolled)
    listener = AsyncListener(handler, handler.channel, source=handler.source, concurrency=10)
    await listener.start()  # validates its subscriptions, then opens the streams
    try:
        yield
    finally:
        await listener.stop()


app = FastAPI(lifespan=lifespan)
```

`AsyncListener` **acks only after success** (nack → redelivery/DLQ). `start()`/`stop()` don't block
the event loop. Only start it when there is at least one `on()` — an empty listener has nothing to
consume.

**Transient-error resilience (native).** Both listeners consume over Pub/Sub **StreamingPull**
(`subscriber.subscribe`): the server streams messages to the client, one stream per subscription, so
**reconnection on transient errors is handled natively by the client** — no manual backoff — and
there is **no head-of-line blocking** between subscriptions (ADR-002, ADR-008/009). Non-transient
errors (permissions, missing resource) still fail fast at `start()`; a stream that dies with a
non-retryable error is logged (`EVENTBUS_STREAM_FAILED`). Flow control bounds in-flight messages
(`max_messages = concurrency`).

**Sync (Celery / Django 3.9+)** — `[pubsub]` (+`[celery]`):

```python
from vaitty_eventbus.runtime import SyncListener

listener = SyncListener(handler, handler.channel, source=handler.source, concurrency=1)
listener.run()  # blocks until stop(); or start()/stop() to run in a daemon thread
```

`SyncListener` opens a Pub/Sub **StreamingPull** per subscription (`subscriber.subscribe`): the
server streams messages to the client with native flow control (`max_messages = concurrency`) and
automatic reconnection — one stream per subscription, so one idle subscription never delays another.

Both are imported lazily (`vaitty_eventbus.runtime.__getattr__`): importing `runtime` does not
parse `aio.py`, so a sync service never drags async code in.

**Ack / nack from a handler.** A handler that **returns** acks the message. A handler that
**raises** nacks it — but an *uncaught* exception is treated as a failure: logged at `ERROR`
(`EVENTBUS_HANDLER_FAILED`) with `str(exc)`. For a nack you *expect* (a dependency is down, or a
business condition like "this tenant isn't configured to receive"), raise **`Nack(reason)`**: the
SDK honors the nack (redelivery/DLQ) but logs it at `INFO` (`EVENTBUS_NACK`) with **only** your
`reason` — a short, safe label you choose, never `str(exc)` or event content. It's the way to say
"nack this on purpose, it's not a bug" without the ERROR noise (ADR-007):

```python
from vaitty_eventbus import Nack


def on_enrolled(event: CorePolicyEnrolled):
    if not tenant_is_receiving(event):  # expected, not a bug
        raise Nack("tenant_not_receiving")  # nack, logged INFO with the reason
    process(event)  # return → ack
    # an unexpected error (e.g. a KeyError) still raises → nack, logged ERROR
```

The `reason` is opaque to the SDK and must not carry event data (keep it a fixed label).

**Raw delivery (`on_raw`) — consume without parsing.** `on()` hands you the **typed DTO** (the SDK
deserializes the body and builds the catalog instance). If you already have your own model and
validation for a topic and don't want the DTO, register `on_raw(pattern, cb)` instead: the listener
**routes by the message's `type` attribute** — it never touches the body — and hands your callback
the raw `RawDelivery` (bytes + read-only `attributes` + runtime signals). You parse **once, your
way** — no double parse (SDK deserializes → you re-parse), no coupling to the catalog DTO (ADR-010):

```python
from vaitty_eventbus import Nack, RawDelivery


def on_enrolled_raw(raw: RawDelivery):
    payload = my_model.parse(raw.raw_data)  # your own parsing, once
    if not tenant_is_receiving(payload):
        raise Nack("tenant_not_receiving")  # same ack contract as on()
    process(payload)  # return → ack


handler.on_raw("core.policy.enrolled", on_enrolled_raw)
```

Same ack/nack contract as `on()` (`return`→ack, `raise`→nack, `Nack(reason)`→intentional nack).
Precedence is **per topic**: if an `on_raw` matches, it **owns** the message and the typed path
(`receive`/`trigger`) is skipped for it; registering both `on()` and `on_raw()` for the same topic
runs **only** the raw one. Subscriptions are derived from the union of `on()` and `on_raw()`, so a
consumer using only `on_raw` still derives and consumes. Inside the callback, `current_delivery()`
exposes the runtime signals and `event_id` (from the attribute); `occurred_at`/`source`/
`schema_version` are `None` (they'd require deserializing).

---

## 🧾 Delivery metadata in the handler (`current_delivery`)

The handler receives the **payload** as the typed DTO. The envelope metadata — who emitted it,
when, under which schema version — is read on the side via `current_delivery()`, without changing
the handler signature. It is populated while a listener is dispatching a message (outside dispatch
it returns `None`):

```python
from vaitty_eventbus.runtime import current_delivery


def on_enrolled(event: CorePolicyEnrolled):
    ctx = current_delivery()  # DeliveryContext for the message in flight (or None)
    log.info(
        "policy=%s occurred_at=%s source=%s schema_v=%s corr=%s attempt=%s msg_id=%s",
        event.policy_number,
        ctx.occurred_at,  # logical event time (from the envelope) — use to order
        ctx.source,  # producing service
        ctx.schema_version,  # payload contract version (forward-compat)
        ctx.correlation_id,  # end-to-end trace id
        ctx.delivery_attempt,  # redelivery counter
        ctx.message_id,  # transport message id (Pub/Sub)
    )
```

`DeliveryContext` carries two groups, both flat and read-only:

- **Envelope metadata:** `event_id`, `occurred_at` (logical event time, for ordering/idempotency —
  ADR-039), `source`, `schema_version`, `correlation_id`, plus `received_at` (when *this* listener
  received it).
- **Channel runtime signals** (neutral, from the channel — not Pub/Sub types; see 002 RN-016):
  `delivery_attempt`, `message_id`, `publish_time`, `attributes` (read-only mapping) and
  `raw_data` (the body **as received**, bytes). `raw_data` lets a consumer compute a content hash
  for dedup without the SDK re-serializing:

```python
import hashlib


def on_enrolled(event: CorePolicyEnrolled):
    ctx = current_delivery()
    dedup_key = ctx.message_id or hashlib.sha256(ctx.raw_data).hexdigest()
    # ... idempotent upsert keyed by dedup_key
```

It is scoped to the message being dispatched (`contextvars`: isolated per task/thread, safe under
concurrency).

---

## 🧭 Topics, subscriptions & provisioning

- **Topic** name: `<domain>.<entity>.<past-verb>` (e.g. `core.policy.enrolled`, `core.visit.confirmed`).
- **Subscription** name: `<service>--<topic>` — the prefix is the consumer's bare service name
  (`EVENTBUS_SOURCE`), e.g. `connect-service--core.policy.enrolled`. The library computes it to
  validate/bind; infra (Terraform) computes the same name to provision it.
- **Provisioning model (ADR-004):** topics and subscriptions are provisioned by **infra**
  (Terraform). The SDK **validates** them at startup (fail-fast with `EVENTBUS_RESOURCE_NOT_FOUND`)
  and **does not create** them. A producer needs only the topic; a consumer needs the topic plus
  its subscription (in Pub/Sub you consume from a subscription, not the topic).
- **Dead-letter / retry / ordering / TTL** are declared as neutral policy; the DLQ topic,
  backoff and retention are infra. See `specs/cross-cutting/delivery-policy.spec.md`.

The full event catalog and envelope live under [`specs/`](specs/) (`events-catalog.asyncapi.yaml`,
`event-envelope.yaml`).

---

## 🔮 Extensibility

1. Subclass `EventChannel`.
2. Register it: `EventHandlerFactory.register_channel("mychannel", MyChannel)`.
3. Select it via `EVENTBUS_CHANNEL=mychannel`.

### Customizing the HTTP transport

For pooled sessions, retries or dynamic auth headers, build `HttpEventChannel` directly:

```python
import requests
from vaitty_eventbus import EventHandler, HttpEventChannel

channel = HttpEventChannel(
    endpoint="http://event-bus-bridge.event-bus-bridge.svc.cluster.local/events",
    session=requests.Session(),
    headers_provider=lambda: {"Authorization": f"Bearer {token_cache.get()}"},
    timeout=5,
)
handler = EventHandler(channel, source="rapihogarweb")
```

`session` accepts anything exposing `.request(method, url, json=, headers=, timeout=)`
(`requests.Session`, `httpx.Client`, …). `headers_provider` is merged on top of `headers` on
every send, so short-lived tokens stay fresh.

---

## 🧪 Development

```bash
git clone git@bitbucket.org:rapihogar/event-bus.git
cd event-bus
python -m venv .venv && source .venv/bin/activate
pip install -e '.[pubsub,fastapi,test,dev,publish]'
pre-commit install
pytest
```

Lint/format are enforced by `ruff` (config in `ruff.toml`); pre-commit runs
`ruff check --fix` and `ruff format` on every commit.

## 🚢 Releasing a new version

Releases are tag-driven: Jenkins publishes to PyPI when a tag matching
`v[0-9]+.[0-9]+.[0-9]+` is pushed and its version equals `pyproject.toml`'s `version`.

```bash
# From a clean `main`, in sync with origin/main:
scripts/release.sh patch          # 0.2.0 → 0.2.1
scripts/release.sh minor          # 0.2.0 → 0.3.0
scripts/release.sh major          # 0.2.0 → 1.0.0
scripts/release.sh 0.2.3          # explicit version
scripts/release.sh patch --dry-run
```

The script bumps `pyproject.toml`, commits `chore(release): vX.Y.Z` on `main`, creates the
annotated tag, pushes both, and back-merges `main` into `develop`.

| Step | Where | Action |
| --- | --- | --- |
| 1 | Feature branch | PR `feature/x` → `develop`. CI: lint + test matrix (Python 3.8–3.14) + coverage gate. |
| 2 | `develop` | PR `develop` → `main`. CI runs again on the merge commit. |
| 3 | `main` (local) | `scripts/release.sh {patch\|minor\|major\|X.Y.Z}` — push is automatic. |
| 4 | Jenkins | The **tag** build runs the Publish stage and uploads to PyPI (trigger the tag build if the multibranch job does not start it on its own). |
