Skip to main content

Topical is a simple event pipeline written in Python.

Project description

topical

Topical is a simple eventing library written in Python.

build

Table of Contents

  1. Developer Requirements
  2. Installation
  3. Usage
    1. Small Example
    2. Using the Event Decorator

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


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for ap-topical, version 0.1.2
Filename, size File type Python version Upload date Hashes
Filename, size ap_topical-0.1.2-py3-none-any.whl (5.0 kB) File type Wheel Python version py3 Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page