Skip to main content
Pre-release

This release is a pre-release and may not be stable for production use.

Python Event Sourcery

A library for event-based systems in Python. For event sourcing, CQRS, and event-driven architectures.

event = EventSourceryIsBorn()

event_store.append(event, stream_id=stream_id)

Under heavy development


Documentation: python-event-sourcery.github.io/python-event-sourcery/


Installation

pip install python-event-sourcery

Easy to setup

Just configure your EventStore instance (or a factory) and call configure_models once. Examples below.

Versatile

Event Sourcery makes no assumptions about your configuration or session management. It's designed to be plugged in into what you already have, without a need to adjust anything. It can be integrated smoothly with thread-scoped SQLAlchemy sessions as well as dependency injection.

NOT opinionated

Although one can easily start with a library, the latter is very customizable and DOES NOT enforce a particular style of using. Also, libraries integrated, i.e. SQLAlchemy and Pydantic are not hardcoded dependencies. They can be replaced with a finite amount of effort.


Use cases & features

Until full documentation is available, you can follow tests

  • Event Sourcing tests
  • Snapshots (for Event Sourcing) tests
  • Event Store - storage for events tests
  • Outbox pattern tests
  • Concurrency control with optimistic locking tests

Quick start

The script is complete, it should run as-is provided FastAPI and library with dependencies are installed. See examples/0-fastapi-integration

from typing import Iterator, Literal
from uuid import uuid4

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, as_declarative

# Import a couple of things from event_sourcery
from event_sourcery import EventStore  # for type annotations
from event_sourcery import (
    configure_models,  # set-up function for library's models
    get_event_store,  # factory function to start quickly
    Event,  # base class for events
)

# Set up your DB and application as you would normally do
engine = create_engine(
    "postgresql://event_sourcery:event_sourcery@localhost:5432/event_sourcery"
)


def db_session() -> Iterator[Session]:
    session = Session(bind=engine)
    try:
        yield session
    finally:
        session.close()


@as_declarative()
class Base:
    pass


app = FastAPI(on_startup=[lambda: Base.metadata.create_all(bind=engine)])

# initialize Event Sourcery models, so they can be handled by SQLAlchemy and e.g. alembic
configure_models(Base)


# Set up factory for event store to be injected into views
def event_store(session: Session = Depends(db_session)) -> EventStore:
    return get_event_store(session)


# Define your event(s) using base-class provided
class CustomerSubscribed(Event):
    plan_id: int
    model: Literal["monthly", "yearly"]


@app.post("/subscription")
def subscribe(
    event_store: EventStore = Depends(event_store),
    session: Session = Depends(db_session),
):
    # Create an event
    event = CustomerSubscribed(plan_id=1, model="yearly")
    # Put it in the event store!
    event_store.append(event, stream_id=uuid4())
    session.commit()

Feature requests / feedback

Anything missing? Create an issue in this repository. Let others upvote

Inspirations

Metadata

Release files for python-event-sourcery 0.5.3a3

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for python-event-sourcery 0.5.3a3
File Size Uploaded
python_event_sourcery-0.5.3a3.tar.gz 216.1 kB Details

Built distribution (wheel)

Table of built distributions (wheels) for python-event-sourcery 0.5.3a3
File Interpreter ABI Platform
python_event_sourcery-0.5.3a3-py3-none-any.whl Python 3 none any Details

Total release size: 278.3 kB

Release files / python_event_sourcery-0.5.3a3.tar.gz

Download URL python_event_sourcery-0.5.3a3.tar.gz
Size 216.1 kB
Tags Source
SHA-256 checksum
How to use checksums
860b67f60bdd0bffbc1352e0933606920ad6782010335e06b4d3467e798ca052
BLAKE2b-256 checksum
How to use checksums
656e5139b7e173b33024916552bb05e15b5d4f714a6cfb96f6c52dc79fd448c5
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via uv/0.12.5 {"installer":{"name":"uv","version":"0.12.5","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}

Release files / python_event_sourcery-0.5.3a3-py3-none-any.whl

Download URL python_event_sourcery-0.5.3a3-py3-none-any.whl
Size 62.2 kB
Tags Python 3
SHA-256 checksum
How to use checksums
3af0b0930d92b853d22f05e653af6a29fe7e1716d93b3e2bef43b1415f0863ee
BLAKE2b-256 checksum
How to use checksums
69c2430cb41b3474a16f82fb27e456981a953ba977d696657819ef8b6f85fc29
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via uv/0.12.5 {"installer":{"name":"uv","version":"0.12.5","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}

Release history Release notifications | RSS feed

This release

0.5.3a3 This release

2 release files

0.5.2

2 release files

0.5.1

2 release files

0.5.0

2 release files

0.4.0

2 release files

0.3.0

2 release files

0.2.0

2 release files

0.1.8

2 release files

0.1.7

2 release files

0.1.6

2 release files

0.1.5

2 release files

0.1.4

2 release files

0.1.3

2 release files

0.1.2

2 release files

0.1.1

2 release files

0.1.0

2 release files

0.0.3

2 release files

0.0.2

2 release files

0.0.1

2 release files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page