Skip to main content

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)

Source distribution for python-event-sourcery 0.5.2
File Size Uploaded
python_event_sourcery-0.5.2.tar.gz 40.5 kB Details

Built distribution (wheel)

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

Release history Release notifications | RSS feed

This release

0.5.2 This release

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