Skip to main content

A toolkit for emitting and handling events, following the CloudEvent spec.

Project description

eventkit-py

ci-badge version-badge

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


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.3.0.tar.gz (11.3 kB view details)

Uploaded Source

Built Distribution

outcome_eventkit-0.3.0-py3-none-any.whl (14.3 kB view details)

Uploaded Python 3

File details

Details for the file outcome-eventkit-0.3.0.tar.gz.

File metadata

  • Download URL: outcome-eventkit-0.3.0.tar.gz
  • Upload date:
  • Size: 11.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.1.4 CPython/3.8.7 Linux/5.4.0-1039-azure

File hashes

Hashes for outcome-eventkit-0.3.0.tar.gz
Algorithm Hash digest
SHA256 376afe985d0e348dbb68e996f5495e98937160d99ae96ce68ab77698c266073d
MD5 ba40c4ef981071bff45e26035c939aff
BLAKE2b-256 2e92cb32d060fb980c8bca148926107b5322c29a2ec32f14bf9adffab36e379a

See more details on using hashes here.

File details

Details for the file outcome_eventkit-0.3.0-py3-none-any.whl.

File metadata

  • Download URL: outcome_eventkit-0.3.0-py3-none-any.whl
  • Upload date:
  • Size: 14.3 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.1.4 CPython/3.8.7 Linux/5.4.0-1039-azure

File hashes

Hashes for outcome_eventkit-0.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 564fbbbfb94d906c84f4871ff7706aead451c5dcb21528a902d91f302961fbf5
MD5 01a89e05d9983e5853f7ee627977f136
BLAKE2b-256 1b7083e56350488f4d8bb66aaf08153937ef169d0f832df489e51fcfbddf4314

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