sikei 0.1.5

This is a fork from Diator, a Python library for implementing CQRS pattern in your applications. The original project is apparently abandoned.

SiKei - CQRS Library for Python

Docs | PyPI

SiKei is a Python library for implementing CQRS pattern in your Python applications forked from Diator (https://akhundmurad.github.io/diator/). It provides a set of abstractions and utilities to help you separate your read and write concerns, allowing for better scalability, performance, and maintainability of your application.

Features :bulb:

• Implements the CQRS pattern.
• Simple, yet flexible API.
• Supports multiple message brokers, such as Redis Pub/Sub and Azure Service Bus.
• Supports various di-frameworks, such as di and rodi.
• Easy to integrate with existing codebases.

Installation :triangular_ruler:

Install the SiKei library with pip

pip install sikei

There are also several installation options:

• To use Redis as Message Broker

  pip install sikei[redis]
  

• Or Azure Service Bus

  pip install sikei[azure]
  

Simple Example :hammer_and_wrench:

Minimal example of sikei usage:

import asyncio
from dataclasses import dataclass, field
from di import Container, bind_by_type
from di.dependent import Dependent
from sikei.events import EventMap, Event, EventEmitter
from sikei.container.di import DIContainer
from sikei.mediator import Mediator
from sikei.requests import Request, RequestHandler, RequestMap


@dataclass(frozen=True, kw_only=True)
class JoinMeetingCommand(Request):
    meeting_id: int
    user_id: int
    is_late: bool = field(default=False)


class JoinMeetingCommandHandler(RequestHandler[JoinMeetingCommand, None]):
    def __init__(self, meeting_api) -> None:
        self._meeting_api = meeting_api
        self._events: list[Event] = []

    @property
    def events(self) -> list[Event]:
        return self._events

    async def handle(self, request: JoinMeetingCommand) -> None:
        self._meeting_api.join(request.meeting_id, request.user_id)
        if request.is_late:
            self._meeting_api.warn(request.user_id)


def setup_di() -> DIContainer:
    external_container = Container()

    external_container.bind(
        bind_by_type(
            Dependent(JoinMeetingCommandHandler, scope="request"),
            JoinMeetingCommandHandler,
        )
    )

    container = DIContainer()
    container.attach_external_container(external_container)

    return container


async def main() -> None:
    container = setup_di()

    request_map = RequestMap()
    request_map.bind(JoinMeetingCommand, JoinMeetingCommandHandler)

    event_emitter = EventEmitter(
        event_map=EventMap(), container=container, message_broker=None
    )

    mediator = Mediator(
        request_map=request_map,
        event_emitter=event_emitter,
        container=container,
    )

    await mediator.send(JoinMeetingCommand(user_id=1, meeting_id=1, is_late=True))


if __name__ == "__main__":
    asyncio.run(main())

Further reading :scroll:

• Udi Dahan - Clarified CQRS
• Martin Fowler - CQRS
• Marting Fowler - What do you mean by “Event-Driven”?
• Vlad Khononov - Learning Domain-Driven Design
• Vaughn Vernon - Really Simple CQRS

License

This project is licensed under the terms of the MIT license.

UV

Using UV to sync, build and publish

uv sync
uv build
uv publish

Inform __token__ and use token created at pypi.org