A toolkit for emitting and handling events, following the CloudEvent spec.
Project description
eventkit-py
A toolkit for emitting and handling events, following the CloudEvent spec.
Installation
poetry add outcome-eventkit
Usage
It's a good idea to be familiar with the CloudEvent v1.0 spec before using this library.
The library implements the following (but is extensible):
Event Example
from outcome.eventkit import CloudEvent, CloudEventData
from outcome.eventkit.formats.json import JSONCloudEventFormat
from outcome.eventkit.protocol_bindings.http import StructuredHTTPBinding, BinaryHTTPBinding
import requests
# Create a payload, specifying how it will be encoded
data = CloudEventData(
data_content_type='application/json',
data={
'hello': 'world'
}
)
# Create an event
event = CloudEvent(type='co.outcome.events.sample', source='example', data=data)
# Create a JSON representation of the event
serialised_event = JSONCloudEventFormat.encode(event)
# Or...
# Create a HTTP message with the event
http_message = BinaryHTTPBinding.to_http(event)
# Or...
# Create a structured HTTP message with the event and a JSON formatter
http_message = StructuredHTTPBinding.to_http(event, JSONCloudEventFormat)
# Post the event somewhere...
requests.post('http://example.org', headers=http_message.headers, data=http_message.body)
Dispatch Example
from outcome.eventkit import dispatch, CloudEvent
# You can register event handlers that will be called
# when a corresponding event is dispatched
# Explicitly
def my_event_handler(event: CloudEvent) -> None:
...
# Register a handler for an event type
dispatch.register_handler('co.outcome.event', my_event_handler)
# Or via a decorator
@dispatch.handles_events('co.outcome.event', 'co.outcome.other-event')
def my_other_handler(event: CloudEvent) -> None:
...
# Then notify all registered handlers of an event
ev = CloudEvent(type='co.outcome.event', source='example')
dispatch.dispatch(ev)
By default, all handlers are stored in a single registry, but you can provide your own registry to the register_handler
/handles_events
/dispatch
methods with the registry
keyword argument.
from outcome.eventkit import dispatch
from collections import defaultdict
my_registry: dispatch.CloudEventHandlerRegistry = defaultdict(list)
@dispatch.handles_events('co.outcome.event', registry=my_registry)
def my_handler(event: CloudEvent) -> None:
...
ev = CloudEvent(type='co.outcome.event', source='example')
dispatch.dispatch(ev, registry=my_registry)
Development
Remember to run ./pre-commit.sh
when you clone the repository.
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 Distribution
outcome-eventkit-0.2.13.tar.gz
(11.3 kB
view hashes)
Built Distribution
Close
Hashes for outcome_eventkit-0.2.13-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 035a27befbaf5099e37d5c4fd04f3237d81543390c6d347edd537e9892715d38 |
|
MD5 | d497586ce750a8d3b5c6260f9eafbec6 |
|
BLAKE2b-256 | da4c19088a0db5627ade014c8d826b204f8de229c566998c8a5c0af2e20f0993 |