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.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | c028be6e7eb5a67df6748388894eb47f02ee3a1a9688ce5ab7e7cc87c32f0e20 |
|
MD5 | 2b0ca535c08f361406a894d19dc1d893 |
|
BLAKE2b-256 | 0fcc903cfa13a0f5fe4e08ec602006902f8d1ddbde537098b515bc9932d13d53 |