python-cqrs 5.3.0 — transactional inbox

Summary

Transactional inbox for a command or an event. Presentation calls InboxRepository.add with a batch of (message_id, message) and commits that session. add returns list[bool] of the same length: True when the row was inserted, False when (message_id, message_name) is already stored, including earlier in the same batch. add does not commit.

class ProcessDeposit(cqrs.Request[None]):
    account_id: str
    amount: str


stored = await inbox.add(
    [
        ("msg-1", ProcessDeposit(account_id="acc-1", amount="10.00")),
        ("msg-1", ProcessDeposit(account_id="acc-1", amount="10.00")),
        ("msg-2", ProcessDeposit(account_id="acc-2", amount="5.00")),
    ],
)
await session.commit()

stored is [True, False, True]. Register each message name on the InboxMessageMap that repository uses before add.

A worker opens its own session, reads get_many, and runs InboxProcessor. process sends a Request through RequestMediator.send and an event through EventMediator.send. It does not commit. The caller commits the business rows and the status together. A handler exception rolls back that row's savepoint, marks the row not_processed, and leaves the rest of the batch in place. After five failures the row leaves get_many.

processor = cqrs.InboxProcessor(
    inbox,
    message_map,
    request_mediator=request_mediator,
)
for row in await inbox.get_many():
    await processor.process(row)
await session.commit()

SqlAlchemyInboxRepository is the [sqlalchemy] extra. SqlModelInboxRepository is the [sqlmodel] extra. Both names stay lazy on cqrs. PostgreSQL and MySQL claim rows with FOR UPDATE SKIP LOCKED.

Install

pip install "python-cqrs[sqlalchemy]"
pip install "python-cqrs[sqlmodel]"

Links

• Docs: https://mkdocs.python-cqrs.dev/latest/inbox/
• Examples: examples/inbox/
• PR: #107