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

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

Built distribution (wheel)

Table of built distributions (wheels) for python-event-sourcery 0.5.3a1
File Interpreter ABI Platform
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

Release history Release notifications | RSS feed

This release

0.5.3a1 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