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.3a1
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.3a1.tar.gz | 40.5 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| python_event_sourcery-0.5.3a1-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 102.6 kB
Release files / python_event_sourcery-0.5.3a1.tar.gz
| Download URL | python_event_sourcery-0.5.3a1.tar.gz |
|---|---|
| Size | 40.5 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
157198af06a597d0074a5ddb0d6916b6caf99a32da9d3a6ec78cd16cd2b99d00
|
|
BLAKE2b-256 checksum How to use checksums |
5cefd18d0facbd60d209fb69aae2ef1ae958d0a88c2b0ec7ffeb450387c4903c
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.1.1 CPython/3.10.21 Linux/6.17.0-1022-azure
|
Release files / python_event_sourcery-0.5.3a1-py3-none-any.whl
| Download URL | python_event_sourcery-0.5.3a1-py3-none-any.whl |
|---|---|
| Size | 62.1 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
1191b34d55dcde5c1c2606fe7c165380c7835655b523a83de3f283cbf73fbfe3
|
|
BLAKE2b-256 checksum How to use checksums |
1a176085f9b8f7a065c5b4884455140fa0e7e71955f858ffcf45662d0be9fe0f
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.1.1 CPython/3.10.21 Linux/6.17.0-1022-azure
|