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.2.tar.gz
(17.9 kB
view hashes)
Built Distribution
Close
Hashes for events_protocol-0.1.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | b23469008d987964ae614993e6420c882291b3e05ddfe40a6e63e118b855d821 |
|
MD5 | b4ffbb17a49fe1f22279f2a4337885bb |
|
BLAKE2b-256 | 5a8c13717e49cc017eb7914f10df57b1257cee3ec0c742e102631d1ea0fcc3d8 |