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.3a4

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.3a4
File Size Uploaded
python_event_sourcery-0.5.3a4.tar.gz 216.1 kB Details

Built distribution (wheel)

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

Total release size: 278.3 kB

Release files / python_event_sourcery-0.5.3a4.tar.gz

Download URL python_event_sourcery-0.5.3a4.tar.gz
Size 216.1 kB
Tags Source
SHA-256 checksum
How to use checksums
557666be107a8803e383ec5bdeb34ad2fb6343ce4d8e992f687ce9c93f93af3c
BLAKE2b-256 checksum
How to use checksums
dab62d55ddc6d44e8a63fa8780ece6b05e46c78cfaa8fc45d91dff781229f8d4
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.3a4-py3-none-any.whl

Download URL python_event_sourcery-0.5.3a4-py3-none-any.whl
Size 62.2 kB
Tags Python 3
SHA-256 checksum
How to use checksums
cb92bb90a58c362fc3a0be0c8ccc3c6248acacafa44aec462e3893bda978d00a
BLAKE2b-256 checksum
How to use checksums
d1dbceb6f72e851801a341cdaa318ff2c9b05a5c1d7f791fd49b24082696d7f1
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.3a4 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