Skip to main content

event bus implementation in python with sync and async support

Project description

event-bus

Run Tests
EventBus implementation in python

Examples

sync

from multi_event_bus import EventBus

eb = EventBus(redis_host="127.0.0.1", redis_port=6379)

eb.dispatch("event-name", payload={"num": 1})

eb.subscribe_to("event-name", consumer_id="consumer-1", offset=0)

event, topic = eb.get(consumer_id="consumer-1")  # Blocking

print(event.payload)  # -> {"num": 1}

async

from multi_event_bus import AsyncEventBus

eb = AsyncEventBus(redis_host="127.0.0.1", redis_port=6379)

eb.dispatch("event-name", payload={"num": 1})

eb.subscribe_to("event-name", consumer_id="consumer-2", offset=0)

event, topic = await eb.get(consumer_id="consumer-2")

print(event.payload)  # -> {"num": 1}

register event schema

from multi_event_bus import EventBus

eb = EventBus(redis_host="127.0.0.1", redis_port=6379)

# Enforce json schema of event
json_schema = {
    "type": "object",
    "properties": {"num": {"type": "string"}}
}
eb.register_event_schema("event-name", schema=json_schema)

eb.dispatch("event-name", payload={"num": "7854"})

eb.subscribe_to("event-name", consumer_id="consumer-3", offset=0)

event, topic = eb.get(consumer_id="consumer-3")  # Blocking

print(event.payload)  # -> {"num": "7854"}

Development

scripts

poetry run run_pytest
poetry run run_flake8
poetry run run_mypy
poetry run run_black

run tests

poetry run test

run all (test and black)

poetry run all

Project details


Download files

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

Source Distribution

multi_event_bus-0.2.0.tar.gz (4.2 kB view details)

Uploaded Source

Built Distribution

multi_event_bus-0.2.0-py3-none-any.whl (4.8 kB view details)

Uploaded Python 3

File details

Details for the file multi_event_bus-0.2.0.tar.gz.

File metadata

  • Download URL: multi_event_bus-0.2.0.tar.gz
  • Upload date:
  • Size: 4.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.1.13 CPython/3.8.10 Linux/5.13.0-39-generic

File hashes

Hashes for multi_event_bus-0.2.0.tar.gz
Algorithm Hash digest
SHA256 504caf6ffa4d51294bab93c644c68ef46a8164fd92f4da2c385ee38b8e22e8bc
MD5 8ce888625cfcfff0fc6767ee1992f8c4
BLAKE2b-256 9814b86413f5248e8e5400fa0b7ada0563b2ca6adeb4c6ff2c7a8eb669c82024

See more details on using hashes here.

File details

Details for the file multi_event_bus-0.2.0-py3-none-any.whl.

File metadata

  • Download URL: multi_event_bus-0.2.0-py3-none-any.whl
  • Upload date:
  • Size: 4.8 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.1.13 CPython/3.8.10 Linux/5.13.0-39-generic

File hashes

Hashes for multi_event_bus-0.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 6a55ae9a7dcdaf294f3f26eb93bf526ed6ef1f76956879a59e79390817c42739
MD5 4ff492e96290cd41c43aae97805f51c2
BLAKE2b-256 b3300923f0bdc90d27f2e59254563cda53b340d08c9b42c5375b99ac261ae30c

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page