pyFly

from pyfly.cqrs import command, query, CommandHandler, QueryHandler

@command(CreateUser)
class CreateUserHandler(CommandHandler[CreateUser, int]):
    def __init__(self, repo: UserRepository) -> None:
        self._repo = repo

    async def handle(self, cmd: CreateUser) -> int:
        return (await self._repo.save(User(name=cmd.name, email=cmd.email))).id

@query(FindUserById, cache_ttl=60)  # caching composes declaratively
class FindUserByIdHandler(QueryHandler[FindUserById, User]):
    def __init__(self, repo: UserRepository) -> None:
        self._repo = repo

    async def handle(self, q: FindUserById) -> User:
        return await self._repo.find_by_id(q.user_id)

# Dispatch — buses are auto-wired by the CqrsAutoConfiguration entry point
user_id = await commands.dispatch(CreateUser(name="Ada", email="ada@example.com"))
user = await queries.dispatch(FindUserById(user_id=user_id))

from pyfly.transactional.saga import saga, saga_step
from pyfly.transactional.saga.core.context import SagaContext

@saga(name="place-order", timeout_ms=30_000)
class PlaceOrderSaga:
    # `compensate=` names a method to invoke if a later step fails.
    @saga_step(id="reserve-inventory", retry=3, backoff_ms=500, compensate="release_inventory")
    async def reserve(self, ctx: SagaContext) -> str:
        return await self._inventory.reserve(ctx.input["order_id"])

    async def release_inventory(self, ctx: SagaContext) -> None:
        await self._inventory.release(ctx.input["order_id"])

    @saga_step(id="charge-payment", depends_on=["reserve-inventory"], compensate="refund_payment")
    async def charge(self, ctx: SagaContext) -> str:
        return await self._payments.charge(ctx.input["order_id"], ctx.input["amount"])

    @saga_step(id="ship-order", depends_on=["charge-payment"])
    async def ship_order(self, ctx: SagaContext) -> None:
        await self._ship.dispatch(ctx.input["order_id"])