| Name | strongbus JSON |
| Version |
0.1.2
JSON |
| download |
| home_page | None |
| Summary | A type-safe event bus library for Python that provides reliable publish-subscribe messaging with automatic memory management, full type safety, and global event subscriptions for cross-cutting concerns like logging. |
| upload_time | 2025-07-14 10:30:37 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.10 |
| license | None |
| keywords |
event-bus
events
pubsub
type-safe
typed
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# StrongBus
A type-safe event bus library for Python that provides reliable publish-subscribe messaging with automatic memory management, full type safety, and global event subscriptions for cross-cutting concerns like logging.
## Features
- **Type Safety**: Full type checking with generics ensures callbacks receive the correct event types
- **Memory Management**: Automatic cleanup of dead references using weak references for methods
- **Subscription Management**: Easy subscription tracking and bulk cleanup via the Enrollment pattern
- **Global Subscriptions**: Subscribe to all events for cross-cutting concerns like logging and monitoring
- **Event Isolation**: Events don't propagate to parent/child types - each event type is handled independently
- **Zero Dependencies**: Pure Python implementation with no external dependencies
## Installation
```bash
pip install strongbus
```
For development:
```bash
pip install -e .
```
## Quick Start
```python
from dataclasses import dataclass
from strongbus import Event, EventBus, Enrollment
# Define your events
@dataclass(frozen=True)
class UserLoginEvent(Event):
username: str
# Create subscribers using Enrollment
class NotificationService(Enrollment):
def __init__(self, event_bus: EventBus):
super().__init__(event_bus)
self.subscribe(UserLoginEvent, self.on_user_login)
def on_user_login(self, event: UserLoginEvent) -> None:
print(f"Welcome {event.username}!")
# Usage
event_bus = EventBus()
service = NotificationService(event_bus)
event_bus.publish(UserLoginEvent(username="Alice"))
# Output: Welcome Alice!
# Cleanup
service.clear() # Automatically unsubscribes from all events
```
## Core Concepts
### Events
Events are simple data classes that inherit from the `Event` base class:
```python
@dataclass(frozen=True)
class OrderCreatedEvent(Event):
order_id: str
customer_id: str
total: float
```
### EventBus
The central hub for publishing and subscribing to events:
```python
event_bus = EventBus()
# Subscribe to events
event_bus.subscribe(OrderCreatedEvent, handle_order)
# Publish events
event_bus.publish(OrderCreatedEvent(
order_id="12345",
customer_id="user123",
total=99.99
))
```
### Enrollment
A base class that simplifies subscription management:
```python
class OrderProcessor(Enrollment):
def __init__(self, event_bus: EventBus):
super().__init__(event_bus)
self.subscribe(OrderCreatedEvent, self.process_order)
self.subscribe(PaymentReceivedEvent, self.confirm_payment)
def process_order(self, event: OrderCreatedEvent) -> None:
# Handle order processing
pass
def confirm_payment(self, event: PaymentReceivedEvent) -> None:
# Handle payment confirmation
pass
```
## Global Event Subscriptions
StrongBus supports global event subscriptions for services that need to receive all events, such as logging or monitoring services:
```python
class LoggerService(Enrollment):
"""Example service that logs all events using global subscription."""
def __init__(self, event_bus: EventBus):
super().__init__(event_bus)
self.subscribe_global(self._log_event)
def _log_event(self, event: Event) -> None:
"""Log any event that occurs."""
event_type = type(event).__name__
print(f"[LOG] {event_type}: {event}")
# Usage
event_bus = EventBus()
logger = LoggerService(event_bus)
# Create other services
notification_service = NotificationService(event_bus)
# All events will be logged automatically
event_bus.publish(UserLoginEvent(username="Alice"))
# Output:
# [LOG] UserLoginEvent: UserLoginEvent(username='Alice')
# Welcome Alice!
event_bus.publish(OrderCreatedEvent(order_id="123", customer_id="user1", total=99.99))
# Output:
# [LOG] OrderCreatedEvent: OrderCreatedEvent(order_id='123', customer_id='user1', total=99.99)
```
Global subscriptions can be managed just like regular subscriptions:
```python
# Unsubscribe from global events
logger.unsubscribe_global(logger._log_event)
# Or clear all subscriptions (including global ones)
logger.clear()
```
## Memory Management
StrongBus automatically manages memory to prevent leaks:
- **Method callbacks** use weak references and are automatically cleaned up when the object is garbage collected
- **Function callbacks** use strong references and persist until explicitly unsubscribed
- **Enrollment pattern** provides easy bulk cleanup with `clear()`
## Testing
### Using tox (recommended)
Install tox with uv support:
```bash
uv tool install tox --with tox-uv
```
Run all tests across multiple Python versions:
```bash
tox
```
### Manual testing
Run the test suite directly:
```bash
python -m unittest src/strongbus/tests.py
```
## Slightly larger example
```python
from dataclasses import dataclass
from strongbus import Event, EventBus, Enrollment
@dataclass(frozen=True)
class UserLoginEvent(Event):
username: str
@dataclass(frozen=True)
class UserLogoutEvent(Event):
username: str
@dataclass(frozen=True)
class DataUpdatedEvent(Event):
data_id: str
new_value: str
@dataclass(frozen=True)
class TestEvent(Event):
message: str
class PackageManager(Enrollment):
def __init__(self, event_bus: EventBus):
super().__init__(event_bus)
# Type-safe subscription - callback must accept UserLoginEvent
self.subscribe(UserLoginEvent, self.on_user_login)
self.subscribe(DataUpdatedEvent, self.on_data_updated)
def on_user_login(self, event: UserLoginEvent) -> None:
# Can access event.username with full type safety
print(f"PackageManager: User {event.username} logged in")
def on_data_updated(self, event: DataUpdatedEvent) -> None:
print(f"PackageManager: Data {event.data_id} updated to {event.new_value}")
class ContainerManager(Enrollment):
def __init__(self, event_bus: EventBus):
super().__init__(event_bus)
self.subscribe(UserLoginEvent, self.on_user_login)
self.subscribe(UserLogoutEvent, self.on_user_logout)
def on_user_login(self, event: UserLoginEvent) -> None:
print(f"ContainerManager: User {event.username} logged in")
def on_user_logout(self, event: UserLogoutEvent) -> None:
print(f"ContainerManager: User {event.username} logged out")
if __name__ == "__main__":
# Usage
event_bus = EventBus()
manager0 = PackageManager(event_bus)
manager1 = ContainerManager(event_bus)
# Publish events - type-safe with proper event objects
event_bus.publish(UserLoginEvent(username="Alice"))
# Output:
# PackageManager: User Alice logged in
# ContainerManager: User Alice logged in
event_bus.publish(UserLogoutEvent(username="Alice"))
# Output:
# ContainerManager: User Alice logged out
event_bus.publish(DataUpdatedEvent(data_id="123", new_value="new data"))
# Output:
# PackageManager: Data 123 updated to new data
# List all available event types
print("\nAvailable event types:")
for event_class in Event.__subclasses__():
print(f" - {event_class.__name__}")
# Cleanup
manager0.clear()
manager1.clear()
```
Raw data
{
"_id": null,
"home_page": null,
"name": "strongbus",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "event-bus, events, pubsub, type-safe, typed",
"author": null,
"author_email": "Marius R\u00e4sener <marius@raesener.de>",
"download_url": "https://files.pythonhosted.org/packages/20/5a/685bf2205d15c4be67c470bc05a6206992655f8b862719948410ea189b55/strongbus-0.1.2.tar.gz",
"platform": null,
"description": "\n# StrongBus\n\nA type-safe event bus library for Python that provides reliable publish-subscribe messaging with automatic memory management, full type safety, and global event subscriptions for cross-cutting concerns like logging.\n\n## Features\n\n- **Type Safety**: Full type checking with generics ensures callbacks receive the correct event types\n- **Memory Management**: Automatic cleanup of dead references using weak references for methods\n- **Subscription Management**: Easy subscription tracking and bulk cleanup via the Enrollment pattern\n- **Global Subscriptions**: Subscribe to all events for cross-cutting concerns like logging and monitoring\n- **Event Isolation**: Events don't propagate to parent/child types - each event type is handled independently\n- **Zero Dependencies**: Pure Python implementation with no external dependencies\n\n## Installation\n\n```bash\npip install strongbus\n```\n\nFor development:\n```bash\npip install -e .\n```\n\n## Quick Start\n\n```python\nfrom dataclasses import dataclass\nfrom strongbus import Event, EventBus, Enrollment\n\n# Define your events\n@dataclass(frozen=True)\nclass UserLoginEvent(Event):\n username: str\n\n# Create subscribers using Enrollment\nclass NotificationService(Enrollment):\n def __init__(self, event_bus: EventBus):\n super().__init__(event_bus)\n self.subscribe(UserLoginEvent, self.on_user_login)\n \n def on_user_login(self, event: UserLoginEvent) -> None:\n print(f\"Welcome {event.username}!\")\n\n# Usage\nevent_bus = EventBus()\nservice = NotificationService(event_bus)\nevent_bus.publish(UserLoginEvent(username=\"Alice\"))\n# Output: Welcome Alice!\n\n# Cleanup\nservice.clear() # Automatically unsubscribes from all events\n```\n\n## Core Concepts\n\n### Events\n\nEvents are simple data classes that inherit from the `Event` base class:\n\n```python\n@dataclass(frozen=True)\nclass OrderCreatedEvent(Event):\n order_id: str\n customer_id: str\n total: float\n```\n\n### EventBus\n\nThe central hub for publishing and subscribing to events:\n\n```python\nevent_bus = EventBus()\n\n# Subscribe to events\nevent_bus.subscribe(OrderCreatedEvent, handle_order)\n\n# Publish events\nevent_bus.publish(OrderCreatedEvent(\n order_id=\"12345\",\n customer_id=\"user123\", \n total=99.99\n))\n```\n\n### Enrollment\n\nA base class that simplifies subscription management:\n\n```python\nclass OrderProcessor(Enrollment):\n def __init__(self, event_bus: EventBus):\n super().__init__(event_bus)\n self.subscribe(OrderCreatedEvent, self.process_order)\n self.subscribe(PaymentReceivedEvent, self.confirm_payment)\n \n def process_order(self, event: OrderCreatedEvent) -> None:\n # Handle order processing\n pass\n \n def confirm_payment(self, event: PaymentReceivedEvent) -> None:\n # Handle payment confirmation\n pass\n```\n\n## Global Event Subscriptions\n\nStrongBus supports global event subscriptions for services that need to receive all events, such as logging or monitoring services:\n\n```python\nclass LoggerService(Enrollment):\n \"\"\"Example service that logs all events using global subscription.\"\"\"\n \n def __init__(self, event_bus: EventBus):\n super().__init__(event_bus)\n self.subscribe_global(self._log_event)\n \n def _log_event(self, event: Event) -> None:\n \"\"\"Log any event that occurs.\"\"\"\n event_type = type(event).__name__\n print(f\"[LOG] {event_type}: {event}\")\n\n# Usage\nevent_bus = EventBus()\nlogger = LoggerService(event_bus)\n\n# Create other services\nnotification_service = NotificationService(event_bus)\n\n# All events will be logged automatically\nevent_bus.publish(UserLoginEvent(username=\"Alice\"))\n# Output: \n# [LOG] UserLoginEvent: UserLoginEvent(username='Alice')\n# Welcome Alice!\n\nevent_bus.publish(OrderCreatedEvent(order_id=\"123\", customer_id=\"user1\", total=99.99))\n# Output:\n# [LOG] OrderCreatedEvent: OrderCreatedEvent(order_id='123', customer_id='user1', total=99.99)\n```\n\nGlobal subscriptions can be managed just like regular subscriptions:\n\n```python\n# Unsubscribe from global events\nlogger.unsubscribe_global(logger._log_event)\n\n# Or clear all subscriptions (including global ones)\nlogger.clear()\n```\n\n## Memory Management\n\nStrongBus automatically manages memory to prevent leaks:\n\n- **Method callbacks** use weak references and are automatically cleaned up when the object is garbage collected\n- **Function callbacks** use strong references and persist until explicitly unsubscribed\n- **Enrollment pattern** provides easy bulk cleanup with `clear()`\n\n## Testing\n\n### Using tox (recommended)\n\nInstall tox with uv support:\n```bash\nuv tool install tox --with tox-uv\n```\n\nRun all tests across multiple Python versions:\n```bash\ntox\n```\n\n### Manual testing\n\nRun the test suite directly:\n```bash\npython -m unittest src/strongbus/tests.py\n```\n\n## Slightly larger example\n```python\nfrom dataclasses import dataclass\n\nfrom strongbus import Event, EventBus, Enrollment\n\n\n@dataclass(frozen=True)\nclass UserLoginEvent(Event):\n username: str\n\n\n@dataclass(frozen=True)\nclass UserLogoutEvent(Event):\n username: str\n\n\n@dataclass(frozen=True)\nclass DataUpdatedEvent(Event):\n data_id: str\n new_value: str\n\n\n@dataclass(frozen=True)\nclass TestEvent(Event):\n message: str\n\n\nclass PackageManager(Enrollment):\n def __init__(self, event_bus: EventBus):\n super().__init__(event_bus)\n # Type-safe subscription - callback must accept UserLoginEvent\n self.subscribe(UserLoginEvent, self.on_user_login)\n self.subscribe(DataUpdatedEvent, self.on_data_updated)\n\n def on_user_login(self, event: UserLoginEvent) -> None:\n # Can access event.username with full type safety\n print(f\"PackageManager: User {event.username} logged in\")\n\n def on_data_updated(self, event: DataUpdatedEvent) -> None:\n print(f\"PackageManager: Data {event.data_id} updated to {event.new_value}\")\n\n\nclass ContainerManager(Enrollment):\n def __init__(self, event_bus: EventBus):\n super().__init__(event_bus)\n self.subscribe(UserLoginEvent, self.on_user_login)\n self.subscribe(UserLogoutEvent, self.on_user_logout)\n\n def on_user_login(self, event: UserLoginEvent) -> None:\n print(f\"ContainerManager: User {event.username} logged in\")\n\n def on_user_logout(self, event: UserLogoutEvent) -> None:\n print(f\"ContainerManager: User {event.username} logged out\")\n\n\nif __name__ == \"__main__\":\n # Usage\n event_bus = EventBus()\n manager0 = PackageManager(event_bus)\n manager1 = ContainerManager(event_bus)\n\n # Publish events - type-safe with proper event objects\n event_bus.publish(UserLoginEvent(username=\"Alice\"))\n # Output:\n # PackageManager: User Alice logged in\n # ContainerManager: User Alice logged in\n\n event_bus.publish(UserLogoutEvent(username=\"Alice\"))\n # Output:\n # ContainerManager: User Alice logged out\n\n event_bus.publish(DataUpdatedEvent(data_id=\"123\", new_value=\"new data\"))\n # Output:\n # PackageManager: Data 123 updated to new data\n\n # List all available event types\n print(\"\\nAvailable event types:\")\n for event_class in Event.__subclasses__():\n print(f\" - {event_class.__name__}\")\n\n # Cleanup\n manager0.clear()\n manager1.clear()\n\n```",
"bugtrack_url": null,
"license": null,
"summary": "A type-safe event bus library for Python that provides reliable publish-subscribe messaging with automatic memory management, full type safety, and global event subscriptions for cross-cutting concerns like logging.",
"version": "0.1.2",
"project_urls": {
"Repository": "https://github.com/elmcrest/strongbus"
},
"split_keywords": [
"event-bus",
" events",
" pubsub",
" type-safe",
" typed"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "ff43938e5bc1841624917453330aa93c96be944f90ec610c3ef77f4e850156c2",
"md5": "cbef24351e0efcf7f08f529888f47e91",
"sha256": "a9ef1b6b548b4e047c58eb4e7ab1590f1be29beacddfb4ca5b88f62957cd0c0e"
},
"downloads": -1,
"filename": "strongbus-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "cbef24351e0efcf7f08f529888f47e91",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 7399,
"upload_time": "2025-07-14T10:30:36",
"upload_time_iso_8601": "2025-07-14T10:30:36.141318Z",
"url": "https://files.pythonhosted.org/packages/ff/43/938e5bc1841624917453330aa93c96be944f90ec610c3ef77f4e850156c2/strongbus-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "205a685bf2205d15c4be67c470bc05a6206992655f8b862719948410ea189b55",
"md5": "b4e9ad4df835268bb3ee5520f8ba665b",
"sha256": "9cae47258f0a36fba218d4f74a80edd0f17155ec68797c7d0bd17179eb732022"
},
"downloads": -1,
"filename": "strongbus-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "b4e9ad4df835268bb3ee5520f8ba665b",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 22323,
"upload_time": "2025-07-14T10:30:37",
"upload_time_iso_8601": "2025-07-14T10:30:37.023002Z",
"url": "https://files.pythonhosted.org/packages/20/5a/685bf2205d15c4be67c470bc05a6206992655f8b862719948410ea189b55/strongbus-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-14 10:30:37",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "elmcrest",
"github_project": "strongbus",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"tox": true,
"lcname": "strongbus"
}
