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.3a5
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| python_event_sourcery-0.5.3a5.tar.gz | 216.8 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| python_event_sourcery-0.5.3a5-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 279.0 kB
Release files / python_event_sourcery-0.5.3a5.tar.gz
| Download URL | python_event_sourcery-0.5.3a5.tar.gz |
|---|---|
| Size | 216.8 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
bf8838e0278fe69659353a0bc0ffcae0d6781d3a0b331af52bdf495fe2726014
|
|
BLAKE2b-256 checksum How to use checksums |
a9abb5241a49ec2e076fdc17b262981a80046bc1db71a14628be20279858bf05
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.12.6 {"installer":{"name":"uv","version":"0.12.6","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.3a5-py3-none-any.whl
| Download URL | python_event_sourcery-0.5.3a5-py3-none-any.whl |
|---|---|
| Size | 62.2 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
2d9d4149127873fc474fd1fe9a44abcb5b42e3309551b92ee51773b27a65acb6
|
|
BLAKE2b-256 checksum How to use checksums |
f54429f0b2f2a580daeaab09c6a85cf053e99ce5815f5201d3ad4ef5183b9a28
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.12.6 {"installer":{"name":"uv","version":"0.12.6","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}
|