Events Protocol’s Documentation¶
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)