# PyEventBus
EventBus implementations in Python. It allows you to easily publish custom domain events and subscribe commands to them.
## Installation
```bash
pip install python3-eventbus
```
## Usage
1. Create the event bus (see "EventBus types" section for more details)
```python
sender_topic = boto3.resource("sns", region_name="us-east-1").Topic(sns_sender_arn)
eventbus = SNSEventBus(sender_topic)
```
2. Define your event
```python
@dataclass
class UserCreatedEvent(DomainEvent):
user_id: str
def to_dict(self) -> dict[str, str]:
return {"user_id": self.user_id}
@classmethod
def from_dict(cls, data: dict[str, str]) -> Self:
return cls(data["user_id"])
```
3. Define the subscriber
```python
@dataclass
class SendWelcomeEmailCommand:
eventbus: EventBus
def __post_init__(self):
self.eventbus.subscribe(
UserCreatedEvent, # The event to subscribe to
lambda event: self.send_welcome_email(event.user_id), # The callback
self.__class__, # The subscriber entity
)
def send_welcome_email(self, user_id: str):
...Business logic...
```
4. Register the subscriber
```python
send_welcome_email_command = SendWelcomeEmailCommand(eventbus) # This will execute the __post_init__ method (dataclass feature)
```
5. Publish the event
```python
eventbus.publish(UserCreatedEvent("user_id"))
```
6. From your controller, rebuild the domain event based on the subscriptions
```python
def lambda_handler(event: dict[str, Any], context: Any):
raw_domain_event = _unwrap_event(event) # Remove event metadata (SQS, SNS, etc.)
domain_event = eventbus.build_event_from_subscriptions(
raw_domain_event["event_type"],
raw_domain_event["event"],
)
```
## EventBus types
### AWS: SNSEventBus
This implementation will forward events to an AWS SNS topic. The topic must be created before the event bus is used.
```python
sender_topic = boto3.resource("sns", region_name="us-east-1").Topic(sns_sender_arn)
eventbus = SNSEventBus(sender_topic)
```
Complete example: [aws_sns.ipynb](./examples/aws_sns.ipynb)
Raw data
{
"_id": null,
"home_page": "",
"name": "python3-eventbus",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.12,<4.0",
"maintainer_email": "",
"keywords": "eventbus,event,bus,python",
"author": "\u00c1lvaro Torres Cogollo",
"author_email": "atorrescogollo@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/b2/18/32bc4bc779033c96902e8c9fe078ca029fdfc2b9554531b24d19bbb290a5/python3_eventbus-0.1.6.tar.gz",
"platform": null,
"description": "# PyEventBus\nEventBus implementations in Python. It allows you to easily publish custom domain events and subscribe commands to them.\n\n## Installation\n```bash\npip install python3-eventbus\n```\n\n## Usage\n1. Create the event bus (see \"EventBus types\" section for more details)\n```python\nsender_topic = boto3.resource(\"sns\", region_name=\"us-east-1\").Topic(sns_sender_arn)\neventbus = SNSEventBus(sender_topic)\n```\n\n2. Define your event\n```python\n@dataclass\nclass UserCreatedEvent(DomainEvent):\n user_id: str\n\n def to_dict(self) -> dict[str, str]:\n return {\"user_id\": self.user_id}\n\n @classmethod\n def from_dict(cls, data: dict[str, str]) -> Self:\n return cls(data[\"user_id\"])\n```\n3. Define the subscriber\n```python\n@dataclass\nclass SendWelcomeEmailCommand:\n eventbus: EventBus\n\n def __post_init__(self):\n self.eventbus.subscribe(\n UserCreatedEvent, # The event to subscribe to\n lambda event: self.send_welcome_email(event.user_id), # The callback\n self.__class__, # The subscriber entity\n )\n\n def send_welcome_email(self, user_id: str):\n ...Business logic...\n```\n4. Register the subscriber\n```python\nsend_welcome_email_command = SendWelcomeEmailCommand(eventbus) # This will execute the __post_init__ method (dataclass feature)\n```\n\n5. Publish the event\n```python\neventbus.publish(UserCreatedEvent(\"user_id\"))\n```\n\n6. From your controller, rebuild the domain event based on the subscriptions\n```python\ndef lambda_handler(event: dict[str, Any], context: Any):\n raw_domain_event = _unwrap_event(event) # Remove event metadata (SQS, SNS, etc.)\n domain_event = eventbus.build_event_from_subscriptions(\n raw_domain_event[\"event_type\"],\n raw_domain_event[\"event\"],\n )\n```\n\n## EventBus types\n### AWS: SNSEventBus\nThis implementation will forward events to an AWS SNS topic. The topic must be created before the event bus is used.\n\n```python\nsender_topic = boto3.resource(\"sns\", region_name=\"us-east-1\").Topic(sns_sender_arn)\neventbus = SNSEventBus(sender_topic)\n```\nComplete example: [aws_sns.ipynb](./examples/aws_sns.ipynb)\n\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Event bus implementation for Python",
"version": "0.1.6",
"project_urls": {
"Issues": "https://github.com/atorrescogollo/pyeventbus/issues"
},
"split_keywords": [
"eventbus",
"event",
"bus",
"python"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "d66ec7ccaeb01a4316a9d64aaec2926f6834cf73ca0ea3a897ddaf3c6b95dadb",
"md5": "3466fc4fcf9a1a3e0319fececebca7ad",
"sha256": "26e561ffc16c4b61f51bc50a73a9b860205b4a6845700ced501672b3f7d3c93f"
},
"downloads": -1,
"filename": "python3_eventbus-0.1.6-py3-none-any.whl",
"has_sig": false,
"md5_digest": "3466fc4fcf9a1a3e0319fececebca7ad",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12,<4.0",
"size": 5593,
"upload_time": "2023-12-28T13:13:55",
"upload_time_iso_8601": "2023-12-28T13:13:55.610512Z",
"url": "https://files.pythonhosted.org/packages/d6/6e/c7ccaeb01a4316a9d64aaec2926f6834cf73ca0ea3a897ddaf3c6b95dadb/python3_eventbus-0.1.6-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "b21832bc4bc779033c96902e8c9fe078ca029fdfc2b9554531b24d19bbb290a5",
"md5": "c8894d745b113ed463b178026c0278cc",
"sha256": "fde57a5b8aa44b38782f5de602844438054dbd40b36568aeec8d8e3589971491"
},
"downloads": -1,
"filename": "python3_eventbus-0.1.6.tar.gz",
"has_sig": false,
"md5_digest": "c8894d745b113ed463b178026c0278cc",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12,<4.0",
"size": 4107,
"upload_time": "2023-12-28T13:13:57",
"upload_time_iso_8601": "2023-12-28T13:13:57.053011Z",
"url": "https://files.pythonhosted.org/packages/b2/18/32bc4bc779033c96902e8c9fe078ca029fdfc2b9554531b24d19bbb290a5/python3_eventbus-0.1.6.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-12-28 13:13:57",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "atorrescogollo",
"github_project": "pyeventbus",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "python3-eventbus"
}
JSON
