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.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 262de55e4365bdd4f5285e1296d032c5894dafb64336e5715745170e583cb0c9 |
|
MD5 | aef9a0319f56cab16f82e64fbd21bf6a |
|
BLAKE2b-256 | a9fbda094fe789ae857dc83decca4c1d0c0157517b129d97f12a3c8d1a020b7b |