vaitty-eventbus 0.0.5

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

To manually install the package, download the vaitty_eventbus-X.X.X-py3-none-any.whl file from the dist directory and run the following command:

pip install vaitty_eventbus-X.X.X-py3-none-any.whl

🛠 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

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)

🌿 Usage

Here's a simple example demonstrating basic usage of vaitty_eventbus:

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 a configured channel
handler.emit("user.login", {"user_id": 42})

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

# Receive an event (maybe some /events endpoint on your app)
event = Event("user.login", {"user_id": 42})

# Call all matched handlers
handler.trigger(event)

🔮 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!