A simple, lightweight event pipeline in Python.
Project description
Topical
A simple, lightweight event pipeline in Python.
Installation
pip install ap-topical
Usage
import logging, asyncio
import ap.topical
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@ap.topical.event('start')
async def start_handler(payload):
"""Do stuff here"""
await ap.topical.publish('end', payload)
@ap.topical.event('start')
async def start_auditor(payload):
logger.info('auditing start event')
@ap.topical.event('end')
async def end_handler(payload):
"""Do stuff here"""
async main():
await ap.topical.publish('start', {})
if __name__ == '__main__':
asyncio.run(main())
Events
Events are strings (or other hashable objects) that are published via ap.topical.publish()
. These are used as both a topic and a specific event.
Subscribing Callbacks to Events
You can subscribe a callback to a given event using ap.topical.subscribe
as in the following example:
async def my_event_handler(payload):
...
ap.topical.subscribe('my-event', my_event_handler)
Additionally, you can use the handy @ap.topical.event()
decorator to do this for you automatically:
@ap.topical.event('my-event')
async def my_event_handler(payload):
...
Publishing Events
Events can be published by providing an event and a payload to pass between event handlers.
await ap.topical.publish('my-event', {})
Multiple Handlers
A given event can have any number of handlers (limited by available compute resources). These will all trigger asynchronously.
@ap.topical.event('start')
async def start_handler(payload):
...
@ap.topical.event('start')
async def start_auditor(payload):
...
await ap.topical.publish('start', {})
This allows you to create robust graphs of event handlers.
Future Plans
- add unsubscribe functionality
- add wildcard event support
- add default event payload support
- add payload validation via msgspec
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.3.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 97885dbace1b26b82b2db7725c0bb2b3a1d5c10b27625210478e63ccfe6800c2 |
|
MD5 | 4399514a78dd81ce7ad3ed0b70986a8f |
|
BLAKE2b-256 | 42c56e1ce07ecf12d27ba377f5b74ae2047a711eacd6f7f8ac1398f6e66e601e |