lexigram-events

Event Sourcing and CQRS engine for Lexigram Framework — domain events, aggregates, and projections.

Overview

CQRS, Event Sourcing, and messaging for Lexigram — command bus, event bus, event store, sagas, and projections. Provides a full CQRS stack: a typed command bus, a pub/sub event bus, an append-only event store (PostgreSQL, SQLite, MongoDB, Redis), saga orchestration, projections, and outbox processing.

Use EventsModule.configure() to register the event system and dispatch commands or subscribe to events via decorators.

Full documentation: docs.lexigram.dev

Install

uv add lexigram-events
# Optional extras
uv add "lexigram-events[postgres,sqlite,mongo,rabbitmq,kafka]"

Quick Start

from lexigram import Application
from lexigram.di.module import Module, module
from lexigram.events import EventsModule, EventsConfig

@module(imports=[EventsModule.configure()])
class AppModule(Module):
    pass

async def main():
    async with Application.boot(modules=[AppModule]) as app:
        # your event sourcing code
        ...

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

Configuration

Zero-config usage: Call EventsModule.configure() with no arguments to use defaults (in-memory event store and bus).

Option 1 — YAML file

# application.yaml
events:
  event_store_backend: postgres
  postgres:
    connection_string: "${DATABASE_URL}"
  event_bus:
    backend: redis
    redis:
      url: "redis://localhost:6379/0"
  outbox:
    enabled: true
    poll_interval: 5

Option 2 — Profiles + Environment Variables (recommended)

export LEX_EVENTS__EVENT_STORE__BACKEND=postgres
export LEX_EVENTS__EVENT_BUS__BACKEND=redis
export LEX_EVENTS__OUTBOX__ENABLED=true

Option 3 — Python

from lexigram.events import EventsModule, EventsConfig
from lexigram.events.types import EventStoreBackend

config = EventsConfig(
    event_store_backend=EventStoreBackend.POSTGRES,
    postgres=PostgresEventStoreConfig(connection_string="${DATABASE_URL}"),
    outbox_enabled=True,
)
EventsModule.configure(config)

Config reference

Module Factory Methods

Key Features

• CommandBus — Typed async command dispatch with middleware
• EventBus — Pub/sub with in-process and adapter-backed delivery
• EventStore — Append-only store (PostgreSQL, SQLite, MongoDB, Redis)
• Saga — Long-running process orchestration with compensating transactions
• Projection — Read-model rebuilding from event streams
• Outbox — Reliable event delivery via transactional outbox pattern
• Schema migration — Versioned event schema evolution

Testing

async with Application.boot(modules=[EventsModule.stub()]) as app:
    # your test code
    ...

Key Source Files