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.2
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.2.tar.gz | 40.5 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| python_event_sourcery-0.5.2-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 102.6 kB
Release files / python_event_sourcery-0.5.2.tar.gz
| Download URL | python_event_sourcery-0.5.2.tar.gz |
|---|---|
| Size | 40.5 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
807a3b67b98ea4e234cb1a0639cc78fe96026310f06a9f7c9cfa15835506b151
|
|
BLAKE2b-256 checksum How to use checksums |
ba2d223671b407f78d1ad0bcc5c6d737d1a729579a36ea43ad4aa394ffb84f53
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.1.1 CPython/3.10.20 Linux/6.14.0-1017-azure
|
Release files / python_event_sourcery-0.5.2-py3-none-any.whl
| Download URL | python_event_sourcery-0.5.2-py3-none-any.whl |
|---|---|
| Size | 62.1 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
528ad8647816670da48989d64e7ac3f79f0b348577fbe959804887162e162449
|
|
BLAKE2b-256 checksum How to use checksums |
6a6b4059238649167056bd6fd70ec9be3ed444a215734cf1c2613e961c6d3c05
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.1.1 CPython/3.10.20 Linux/6.14.0-1017-azure
|