Events Protocol’s Documentation

PyPI version Actions Status Documentation Status License Code style: black PyPI - Python Version

Guiabolso Logo

Library to be a Client and Server using Guiabolso’s Events Protocol Specification, the especification of this implementation can be found here

Installation

To install stable version, just download it from PyPI:

pip install events-protocol

To install from source code execute the following command:

pip install git+https://github.com/GuiaBolso/events-protocol-python#egg=events-protocol

Basic Usage

Client

The essencial information to send an event is:

Field

Description

url

URL that was exposed from Event Server

name

Event name registred on Event Server source

version

Event’s version

payload

Payload with necessery event information

With this informatons we can send an event.

Instantiate the client:

from events_protocol.client import EventClient

client = EventClient(url="http://example.com/events/")

Send event:

# Exemplo passando apenas as informações essenciais
response = client.send_event(
    name="event:example",
    version=1,
    payload={
        "example": "example"
    },
)

Or you can send the event passing all of the informatons:

response = client.send_event(
    name="event:example",
    version=1,
    id="9230c47c-3bcf-11ea-b77f-2e728ce88125",
    flow_id="a47830ca-3bcf-11ea-a232-2e728ce88125",
    payload={
        "example": "example"
    },
    identity={
        "userId": "USER_ID",
    },
    metadata={
        "date": "00-00-0000",
    },
    timeout=1000,
)

Server

An Event Server is composed by:

Component

Description

handler

Class that will receive and process the event

register

Class that will register the handler to Event Discovery

EventSchema

Event Schema will be accepted in the payload attribute

Example:

from events_protocol.server.handler.event_handler_registry import EventRegister
from events_protocol.core.builder import EventBuilder, Event
from events_protocol.core.model.base import CamelPydanticMixin
from events_protocol.core.model.event import Event, ResponseEvent
from events_protocol.server.handler.event_handler import EventHandler
from events_protocol.server.parser.event_processor import EventProcessor


class MyEventSchema(CamelPydanticMixin):
    example: str


class MyHandler(EventHandler):
    _SCHEMA = MyEventSchema

    @classmethod
    def handle(cls, event: Event) -> ResponseEvent:
        payload = cls.parse_event(event)
        response = {"MyEventPayload": payload.example}
        return EventBuilder.response_for(event, response)


class MyEventRegister(EventRegister):
    event_name = "get:event:example"
    event_version = 1
    event_handler = MyHandler


MyEventRegister.register_event()

event_input = Event(
    name="get:event:example",
    version=1,
    id="9230c47c-3bcf-11ea-b77f-2e728ce88125",
    flow_id="a47830ca-3bcf-11ea-a232-2e728ce88125",
    payload={
        "example": "example"
    },
    identity={
        "userId": "USER_ID",
    },
    metadata={
        "date": "00-00-0000",
    },
)
input_body = event_input.to_json()

response = EventProcessor.process_event(input_body)

Content