Topical is a simple event pipeline written in Python.
Project description
topical
Topical is a simple eventing library written in Python.
Table of Contents
Developer Requirements
- Python 3.7 or higher
Installation
pip install ap-topical
Usage
Topical supports both synchronous and asynchronous callbacks. This documentation only details the async/await calls. For more in-depth usage instructions, please consult the wiki.
First, you will need an event handler. This function requires a payload
object as its only parameter:
async def event_handler_one(payload):
pass
async def event_handler_two(payload):
pass
Next, you will need to configure your event pipeline.
topical.subscribe('event-one', event_handler_one)
topical.subscribe('event-two', event_handler_two)
Next, you will need to publish your first event.
payload = TopicalEventPayload()
await topical.publish_async('event-one', payload)
To continue your event chain, you will need to publish an event from an existing event handler:
async def event_handler_one(payload):
# do something with your event payload
await topical.publish_async('event_two', payload)
Small Example
import asyncio
from ap.topical import topical
from ap.topical.topical_event_payload import TopicalEventPayload
async def event_handler_one(payload):
print(f'inside event_handler_one for {payload.idempotency_token}')
print('sleeping for 5 seconds')
await asyncio.sleep(5)
await topical.publish_async('event-two', payload)
async def event_handler_two(payload):
print(f'inside event_handler_two for {payload.idempotency_token}')
print('sleeping for 5 seconds')
await asyncio.sleep(5)
print('done!')
async def main():
topical.subscribe('event-one', event_handler_one)
topical.subscribe('event-two', event_handler_two)
payloads = [TopicalEventPayload() for _ in range(3)]
await asyncio.gather(*[topical.publish_async('event-one', payload) for payload in payloads])
asyncio.run(main())
# Output
# (topical) λ python info.py
# inside event_handler_one for 3744628e-fb1c-4c60-a43a-175b1a09b0fd
# sleeping for 5 seconds
# inside event_handler_one for 64a546dc-46fd-4c23-9adc-501194376ea6
# sleeping for 5 seconds
# inside event_handler_one for b470535f-f65a-4c22-b764-6fe8379a0388
# sleeping for 5 seconds
# inside event_handler_two for 3744628e-fb1c-4c60-a43a-175b1a09b0fd
# sleeping for 5 seconds
# inside event_handler_two for 64a546dc-46fd-4c23-9adc-501194376ea6
# sleeping for 5 seconds
# inside event_handler_two for b470535f-f65a-4c22-b764-6fe8379a0388
# sleeping for 5 seconds
# done!
# done!
# done!
Using the Event Decorator
Manually subscribing to a lot of events could be an issue. Topical ships with a decorator that allows you to decorate any function as an event handler.
import asyncio
from ap.topical import topical
@topical.event('event-one')
async def event_handler_one(payload):
print('in event handler')
async def main():
await asyncio.gather(*[topical.publish_async('event-one', {}) for _ in range(3)])
asyncio.run(main())
# Output
# (topical) λ python info.py
# in event handler
# in event handler
# in event handler
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
Built Distribution
Hashes for ap_topical-0.2.5-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 5c49ec3ce1a3afc1a99523bf9728c3449901a655bd78667f8a4ff1522a8df079 |
|
MD5 | da1b3d34703bbb5c602630cba76bf586 |
|
BLAKE2b-256 | 1199926a016b8679485736aff9774c1143a40f6e046e42a07ea29f15aff9c7d0 |