Library to be a Client and Server using event protocol
Project description
events-protocol
Como usar
Client
As informações essenciais para enviar o evento são: url, name, version e payload.
Apenas com estas informações já é possivel enviar um evento.
from events_protocol.client import EventClient
# Instancia o client
client = EventClient(url="http://example.com/events/")
# Exemplo passando apenas as informações essenciais
response = client.send_event(
name="event:example",
version=1,
payload={
"example": "example"
},
)
# Exemplo passando todas as informações
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
Um server é composto por handler, register e EventSchema.
Abaixo se encontra um exemplo de utilização.
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()
## Apos todos eventos registrados, registre uma rota "/events" no seu framework web de preferência e processe o body utilizando o seguinte comando
response = EventProcessor.process_event(input_body)
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
events-protocol-0.1.4.tar.gz
(18.1 kB
view hashes)
Built Distribution
Close
Hashes for events_protocol-0.1.4-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 5898aa5d5b18aecf638873879c63e922ae94c80bc348116df50eeec55a35f8e2 |
|
MD5 | 86f46b9be078d17d2e497c7181a50da7 |
|
BLAKE2b-256 | d892f0544963dc8552eb6e8d7eba0eb9828fe8b4fae73a107e4ec8a272c409dd |