vaitty-eventbus 0.1.0

A minimal, internal Python library for decoupled event communication using channels and wildcard-capable handlers.

vaitty_eventbus

A minimal, internal Python library for decoupled event communication using channels and wildcard-capable handlers.

---

🚀 Features

• 🔌 Pluggable event channels (console, file, or custom)
• 🎯 Wildcard-matching event handlers (user.*, *)
• 🔁 Emits and triggers events independently
• 🧼 No third-party dependencies
• 🧩 Fully configurable via Python dict or environment variables

---

📦 Installation

pip install vaitty-eventbus

Pin a specific version in your project's requirements:

vaitty-eventbus==0.0.5

🛠 Configuration

You can configure vaitty_eventbus through a dictionary or environment variables:

• EVENTBUS_CHANNEL: Determines the channel type to use: console, file, http, or a registered custom type. Default: console
• EVENTBUS_FILE: For the file channel, the path to the log file. Default: events.log
• EVENTBUS_HTTP_ENDPOINT: For the http channel, the target HTTP endpoint. Default: http://example.com/events
• EVENTBUS_HTTP_METHOD: For the http channel, the HTTP method used to dispatch events. Default: POST
• EVENTBUS_HTTP_TIMEOUT: For the http channel, per-request timeout in seconds. Default: no timeout

Example using environment variables:

export EVENTBUS_CHANNEL=file
export EVENTBUS_FILE=/tmp/my-events.log

Example using a configuration dictionary:

from vaitty_eventbus import EventHandlerFactory

config = {
    "EVENTBUS_CHANNEL": "http",
    "EVENTBUS_HTTP_ENDPOINT": "http://api.my-events.com/ingest"
}
handler = EventHandlerFactory.from_config(config=config)

Customizing the HTTP transport

For pooled sessions, retries, dynamic auth headers or custom HTTP clients, instantiate HttpEventChannel directly and wrap it in an EventHandler:

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

from vaitty_eventbus import EventHandler, HttpEventChannel

session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=Retry(total=3, backoff_factor=0.5)))

channel = HttpEventChannel(
    endpoint="https://ingest.internal/events",
    session=session,
    method="POST",
    headers={"X-Service": "ai-gateway"},
    headers_provider=lambda: {"Authorization": f"Bearer {token_cache.get()}"},
    timeout=5,
)
handler = EventHandler(channel)

session accepts any object exposing .request(method, url, json=, headers=, timeout=) — requests.Session, httpx.Client and similar work without changes. headers_provider is invoked on every send and merged on top of headers, so short-lived tokens stay fresh.

🌿 Usage

vaitty_eventbus keeps two flows independent:

• Emit pushes an event out through the configured channel (console / file / HTTP / custom).
• Subscribe + dispatch lets your app react to incoming events. The library only matches callbacks against a topic; your code runs them, so the library stays neutral on sync vs async runtimes (Django 2.2 / py3.9 monoliths and FastAPI services use the same API).

from typing import Any, Dict

from vaitty_eventbus import EventHandlerFactory, Event

def handle_all(payload: Dict[str, Any]):
    print(f"[*] Received: {payload}")

def handle_user(payload: Dict[str, Any]):
    print(f"[user.*] User event: {payload}")

def handle_login(payload: Dict[str, Any]):
    print(f"[user.login] Specific login: {payload['user_id']}")

handler = EventHandlerFactory.from_config()

# Send an event using the configured channel
handler.emit("user.login", {"user_id": 42})

# Register handlers for each topic or use wildcards
handler.on("*", handle_all)
handler.on("user.*", handle_user)
handler.on("user.login", handle_login)

Dispatching incoming events

When your app receives an event (e.g. from an /events HTTP endpoint), build an Event and iterate the matching callbacks. The library returns a lazy generator — each callback yielded only once even if multiple patterns match.

Sync runtime (Django views, scripts, workers):

event = Event("user.login", {"user_id": 42})

for callback in handler.get_matching_callbacks(event):
    callback(event.payload)

Async runtime (FastAPI, asyncio workers):

import asyncio
import inspect

async def dispatch(handler, event):
    for callback in handler.get_matching_callbacks(event):
        if inspect.iscoroutinefunction(callback):
            asyncio.create_task(callback(event.payload))
        else:
            callback(event.payload)

await dispatch(handler, Event("user.login", {"user_id": 42}))

The library does not import asyncio itself, so you choose the dispatch model that fits your stack.

🔮 Extensibility

The library is designed to be easily extended:

1. Add new EventChannel subclasses.
2. Register them with EventHandlerFactory.register_channel.
3. Configure their behavior through the dictionary or environment variables.

This makes it simple to integrate vaitty_eventbus into diverse workflows and systems!

🧪 Development

git clone [e-mail]:rapihogar/event-bus.git
cd event-bus
python -m venv .venv && source .venv/bin/activate
pip install -e '.[test,dev,publish]'
pre-commit install
pytest

Linting and formatting are enforced by ruff (config in pyproject.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.

The release script handles the full cut from main:

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

What it does atomically:

1. Verifies branch is main, working tree is clean, and main is in sync with origin/main.
2. Verifies the target tag does not already exist locally or on origin.
3. Bumps pyproject.toml and prints the planned actions for confirmation.
4. Commits chore(release): vX.Y.Z on main, creates annotated tag vX.Y.Z, pushes both.
5. Back-merges main into develop and pushes — keeps both branches aligned so the next develop → main PR doesn't fight over the version line.

Jenkins picks up the tag, builds the wheel + sdist, runs twine check, and uploads to public PyPI. The build aborts if the tag does not match the package version, so a mismatched release is structurally impossible.

Full release flow

Step	Where	Action
1	Feature branch	Open PR feature/x → develop. CI runs (lint + test matrix on Python 3.9–3.14 + coverage gate).
2	develop	Open PR develop → main. CI runs again on the merge commit.
3	main (local)	Run scripts/release.sh {patch|minor|major|X.Y.Z}. Push happens automatically.
4	Jenkins	Tag build runs the Publish stage, uploads to PyPI.

There is no separate version-bump PR. The bump lives in the same commit as the release tag, on main, written by the script.