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)
| File | Size | Uploaded | |
|---|---|---|---|
| python_event_sourcery-0.5.3a3.tar.gz | 216.1 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| 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}
|