Source code for eventscore.ext.kafka.serializers

import json

from eventscore.core.abstract import IEventSerializer
from eventscore.core.types import Event, EventDict


[docs] class KafkaEventSerializer(IEventSerializer[EventDict, bytes]):
[docs] def encode(self, event: Event) -> bytes: return json.dumps(event.asdict()).encode()
[docs] def decode(self, event: EventDict) -> Event: return Event.fromdict(event)