### Basic Application Setup with SQLAlchemy Persistence (Python)

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Demonstrates configuring an event-sourced application to use SQLAlchemy persistence by setting environment variables for the persistence module and database URL. It defines a Dog aggregate and a TrainingSchool application, then shows how to register a dog, add tricks, and retrieve them.

```python
import os
from eventsourcing.application import Application
from eventsourcing.domain import Aggregate, event
from uuid import uuid5, NAMESPACE_URL

# Configure SQLAlchemy persistence
os.environ['PERSISTENCE_MODULE'] = 'eventsourcing_sqlalchemy'
os.environ['SQLALCHEMY_URL'] = 'sqlite:///:memory:'

# Define a domain aggregate
class Dog(Aggregate):
    @event('Registered')
    def __init__(self, name):
        self.name = name
        self.tricks = []

    @staticmethod
    def create_id(name):
        return uuid5(NAMESPACE_URL, f'/dogs/{name}')

    @event('TrickAdded')
    def add_trick(self, trick):
        self.tricks.append(trick)

# Define an application
class TrainingSchool(Application):
    def register(self, name):
        dog = Dog(name)
        self.save(dog)

    def add_trick(self, name, trick):
        dog = self.repository.get(Dog.create_id(name))
        dog.add_trick(trick)
        self.save(dog)

    def get_tricks(self, name):
        dog = self.repository.get(Dog.create_id(name))
        return dog.tricks

# Use the application
school = TrainingSchool()
school.register('Fido')
school.add_trick('Fido', 'roll over')
school.add_trick('Fido', 'play dead')

tricks = school.get_tricks('Fido')
print(tricks)  # Output: ['roll over', 'play dead']
```

--------------------------------

### Paginate Notifications with SQLAlchemyProcessRecorder

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Shows how to paginate through recorded notifications using 'start' and 'limit' parameters. This allows for efficient retrieval of event data in chunks.

```python
from eventsourcing_sqlalchemy.recorders import SQLAlchemyProcessRecorder
from eventsourcing_sqlalchemy.datastore import SQLAlchemyDatastore

# Assuming 'recorder' is an instance of SQLAlchemyProcessRecorder
datastore = SQLAlchemyDatastore(url='sqlite:///:memory:')
recorder = SQLAlchemyProcessRecorder(
    datastore=datastore,
    events_table_name='process_events',
    tracking_table_name='notification_tracking',
    schema_name=None
)
recorder.create_table()

notifications = recorder.select_notifications(start=1, limit=10, stop=1)
print(f"First notification topic: {notifications[0].topic}")
```

--------------------------------

### SQLAlchemyDatastore Initialization and Usage (Python)

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Shows different ways to initialize and use the SQLAlchemyDatastore. Examples include creating a datastore with an in-memory SQLite URL, checking if it's an in-memory SQLite database, creating a datastore with a pre-configured session maker, and initializing one with a scoped session for web applications.

```python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from eventsourcing_sqlalchemy.datastore import SQLAlchemyDatastore

# Create a datastore with a URL
datastore = SQLAlchemyDatastore(url='sqlite:///:memory:', autoflush=True)

# Check database type
print(f"Is SQLite in-memory: {datastore.is_sqlite_in_memory_db}")
# Output: Is SQLite in-memory: True

# Create a datastore with an existing session maker
engine = create_engine('postgresql://user:password@localhost/mydb')
session_maker = sessionmaker(bind=engine)
datastore = SQLAlchemyDatastore(session_maker=session_maker)

# Create a datastore with a scoped session (for web applications)
scoped = scoped_session(sessionmaker(bind=engine))
datastore = SQLAlchemyDatastore(session=scoped)
print(f"Using scoped session: {datastore.scoped_session is not None}")
# Output: Using scoped session: True
```

--------------------------------

### Define Domain Models for Event Sourcing

Source: https://github.com/pyeventsourcing/eventsourcing-sqlalchemy/blob/main/README.md

This Python code defines a `TrainingSchool` application and a `Dog` aggregate using the `eventsourcing` library. It demonstrates how to create events, manage aggregate state, and use static methods for ID generation. This serves as a foundational example for building event-sourced applications.

```python
from eventsourcing.application import Application
from eventsourcing.domain import Aggregate, event
from uuid import uuid5, NAMESPACE_URL


class TrainingSchool(Application):
    def register(self, name):
        dog = Dog(name)
        self.save(dog)

    def add_trick(self, name, trick):
        dog = self.repository.get(Dog.create_id(name))
        dog.add_trick(trick)
        self.save(dog)

    def get_tricks(self, name):
        dog = self.repository.get(Dog.create_id(name))
        return dog.tricks


class Dog(Aggregate):
    @event('Registered')
    def __init__(self, name):
        self.name = name
        self.tricks = []

    @staticmethod
    def create_id(name):
        return uuid5(NAMESPACE_URL, f'/dogs/{name}')

    @event('TrickAdded')
    def add_trick(self, trick):
        self.tricks.append(trick)
```

--------------------------------

### Configure PostgreSQL with Custom Schema

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Illustrates how to configure a PostgreSQL database to use a specific schema for storing event sourcing data. This is useful for multi-tenant applications or isolating data. The example shows setting the `SQLALCHEMY_SCHEMA` environment variable.

```python
import os
from eventsourcing.application import Application

# Configure PostgreSQL with schema
os.environ['PERSISTENCE_MODULE'] = 'eventsourcing_sqlalchemy'
os.environ['SQLALCHEMY_URL'] = 'postgresql://user:password@localhost:5432/mydb'
os.environ['SQLALCHEMY_SCHEMA'] = 'tenant_001'  # Events stored in tenant_001 schema

app = Application()

# Tables created:
# - tenant_001.stored_events
# - tenant_001.stored_snapshots (if snapshots enabled)
# - tenant_001.notification_tracking (for process applications)

print(f"Schema: {app.factory._schema_name}")
# Output: Schema: tenant_001
```

--------------------------------

### Configure Event Sourcing with SQLAlchemy

Source: https://github.com/pyeventsourcing/eventsourcing-sqlalchemy/blob/main/README.md

This Python snippet configures the `eventsourcing_sqlalchemy` persistence module by setting environment variables. `PERSISTENCE_MODULE` is set to `'eventsourcing_sqlalchemy'`, and `SQLALCHEMY_URL` is set to an in-memory SQLite database URL. This setup is crucial for enabling SQLAlchemy as the backend for event storage.

```python
import os

os.environ['PERSISTENCE_MODULE'] = 'eventsourcing_sqlalchemy'
os.environ['SQLALCHEMY_URL'] = 'sqlite:///:memory:'
```

--------------------------------

### Google Cloud SQL Connection Creator for Eventsourcing

Source: https://github.com/pyeventsourcing/eventsourcing-sqlalchemy/blob/main/README.md

Configures the eventsourcing library to use Google Cloud SQL for database connections via the Cloud SQL Python Connector. It involves installing the connector, defining a `getconn()` function to establish the connection, and setting specific environment variables to point the persistence module and connection creator topic.

```python
from google.cloud.sql.connector import Connector
from eventsourcing.utils import get_topic
import os

connector = Connector()

def get_google_cloud_sql_conn():
    return connector.connect(
        "project:region:instance",
        "pg8000",
        user="postgres-iam-user@gmail.com",
        db="my-db-name",
        enable_iam_auth=True,
   )

os.environ['PERSISTENCE_MODULE'] = 'eventsourcing_sqlalchemy'
os.environ['SQLALCHEMY_URL'] = 'postgresql+pg8000://'
os.environ['SQLALCHEMY_CONNECTION_CREATOR_TOPIC'] = get_topic(get_google_cloud_sql_conn)
```

--------------------------------

### Define SQLAlchemy Database Models

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Provides the base SQLAlchemy models for event sourcing persistence. It includes `StoredEventRecord` for domain events, `SnapshotRecord` for aggregate snapshots, and `NotificationTrackingRecord` for tracking processed notifications. The example shows how to create the database engine and tables.

```python
from eventsourcing_sqlalchemy.models import (
    StoredEventRecord,
    SnapshotRecord,
    NotificationTrackingRecord,
    Base
)
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Create engine and tables
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

# StoredEventRecord columns:
# - id: BigInteger (auto-increment, primary key)
# - originator_id: UUID (aggregate ID)
# - originator_version: BigInteger (version number)
# - topic: Text (event class path)
# - state: LargeBinary (serialized event data)

# SnapshotRecord columns:
# - originator_id: UUID (primary key)
# - originator_version: BigInteger (primary key)
# - topic: Text
# - state: LargeBinary

# NotificationTrackingRecord columns:
# - application_name: String(32) (primary key)
# - notification_id: BigInteger (primary key)

Session = sessionmaker(bind=engine)
session = Session()

# Query example (typically done through recorders)
events = session.query(StoredEventRecord).all()
print(f"Total events: {len(events)}")
# Output: Total events: 0
```

--------------------------------

### Get Max Tracking ID with SQLAlchemyProcessRecorder

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Shows how to retrieve the maximum notification ID that has been tracked for a specific application. This is useful for determining the next notification to process.

```python
from eventsourcing.persistence import StoredEvent, Tracking
from eventsourcing_sqlalchemy.datastore import SQLAlchemyDatastore
from eventsourcing_sqlalchemy.recorders import SQLAlchemyProcessRecorder
from uuid import uuid4

datastore = SQLAlchemyDatastore(url='sqlite:///:memory:')
recorder = SQLAlchemyProcessRecorder(
    datastore=datastore,
    events_table_name='process_events',
    tracking_table_name='notification_tracking',
    schema_name=None
)
recorder.create_table()

originator_id = uuid4()
events = [
    StoredEvent(
        originator_id=originator_id,
        originator_version=1,
        topic='order:Created',
        state=b'{"order_id": "12345"}'
    ),
]
tracking = Tracking(application_name='upstream_app', notification_id=5)
recorder.insert_events(events, tracking=tracking)

max_tracking = recorder.max_tracking_id('upstream_app')
print(f"Max tracking ID: {max_tracking}")
```

--------------------------------

### Instantiate and Use Event Sourcing Application

Source: https://github.com/pyeventsourcing/eventsourcing-sqlalchemy/blob/main/README.md

This Python code demonstrates how to instantiate and use an event sourcing application after configuring it with SQLAlchemy. It creates a `TrainingSchool` instance, registers a dog, adds tricks to the dog's record, and then retrieves the list of tricks, asserting the expected outcome. This showcases the end-to-end usage of the configured application.

```python
school = TrainingSchool()
school.register('Fido')
school.add_trick('Fido', 'roll over')
school.add_trick('Fido', 'play dead')
tricks = school.get_tricks('Fido')
assert tricks == ['roll over', 'play dead']
```

--------------------------------

### Initialize SQLAlchemyProcessRecorder

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Demonstrates the initialization of SQLAlchemyProcessRecorder with a SQLAlchemyDatastore. It includes creating the necessary tables for events and tracking.

```python
from eventsourcing.persistence import StoredEvent, Tracking
from eventsourcing_sqlalchemy.datastore import SQLAlchemyDatastore
from eventsourcing_sqlalchemy.recorders import SQLAlchemyProcessRecorder
from uuid import uuid4

datastore = SQLAlchemyDatastore(url='sqlite:///:memory:')
recorder = SQLAlchemyProcessRecorder(
    datastore=datastore,
    events_table_name='process_events',
    tracking_table_name='notification_tracking',
    schema_name=None
)
recorder.create_table()
```

--------------------------------

### SQLAlchemyFactory Configuration with Custom Environment (Python)

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Illustrates how to create an Application instance using a custom Environment object with SQLAlchemy persistence. It sets environment variables for the persistence module, database URL, an optional schema, and table creation, then accesses the SQLAlchemyFactory and its datastore.

```python
from eventsourcing.application import Application
from eventsourcing.utils import Environment
from eventsourcing_sqlalchemy.factory import SQLAlchemyFactory

# Create an application with custom environment settings
env = Environment("MyApp")
env['PERSISTENCE_MODULE'] = 'eventsourcing_sqlalchemy'
env['SQLALCHEMY_URL'] = 'postgresql://user:password@localhost:5432/mydb'
env['SQLALCHEMY_SCHEMA'] = 'events'  # Optional: use a database schema
env['CREATE_TABLE'] = 'yes'  # Auto-create tables (default: yes)

app = Application(env=env)

# Access the factory and datastore
assert isinstance(app.factory, SQLAlchemyFactory)
print(f"Database engine: {app.factory.datastore.engine}")
# Output: Database engine: Engine(postgresql://user:***@localhost:5432/mydb)
```

--------------------------------

### Integrate with Flask-SQLAlchemy - Python

Source: https://github.com/pyeventsourcing/eventsourcing-sqlalchemy/blob/main/README.md

Illustrates how to integrate pyeventsourcing with Flask-SQLAlchemy by adapting its SQLAlchemy session. This involves defining a Flask app, configuring the database URI, and creating a SQLAlchemy instance with a declarative base. The Flask-SQLAlchemy session can then be used similarly to a standard scoped session.

```python
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
try:
    from sqlalchemy.orm import declarative_base  # type: ignore
except ImportError:
    from sqlalchemy.ext.declarative import declarative_base

# Define a Flask app.
flask_app = Flask(__name__)
flask_app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'

# Integration between Flask and SQLAlchemy.
Base = declarative_base()
db = SQLAlchemy(flask_app, model_class=Base)
```

--------------------------------

### Insert Events with Tracking using SQLAlchemyProcessRecorder

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Illustrates how to insert events along with tracking information using SQLAlchemyProcessRecorder. This is crucial for implementing exactly-once processing semantics.

```python
from eventsourcing.persistence import StoredEvent, Tracking
from eventsourcing_sqlalchemy.datastore import SQLAlchemyDatastore
from eventsourcing_sqlalchemy.recorders import SQLAlchemyProcessRecorder
from uuid import uuid4

datastore = SQLAlchemyDatastore(url='sqlite:///:memory:')
recorder = SQLAlchemyProcessRecorder(
    datastore=datastore,
    events_table_name='process_events',
    tracking_table_name='notification_tracking',
    schema_name=None
)
recorder.create_table()

originator_id = uuid4()
events = [
    StoredEvent(
        originator_id=originator_id,
        originator_version=1,
        topic='order:Created',
        state=b'{"order_id": "12345"}'
    ),
]

tracking = Tracking(application_name='upstream_app', notification_id=5)
recorder.insert_events(events, tracking=tracking)
```

--------------------------------

### Verify Aggregate Persistence with SQLAlchemy

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Demonstrates how to retrieve an aggregate from the repository after it has been saved and committed to the database. It also shows how a transaction rollback prevents an aggregate from being persisted.

```python
with db(commit_on_exit=True):
    retrieved = es_app.repository.get(aggregate.id)
    print(f"Aggregate persisted: {retrieved.id == aggregate.id}")
    # Output: Aggregate persisted: True

# Failed request - not committed
try:
    with db(commit_on_exit=True):
        new_aggregate = Aggregate()
        es_app.save(new_aggregate)
        raise ValueError("Request failed")
except ValueError:
    pass

# Verify aggregate not persisted
with db(commit_on_exit=True):
    try:
        es_app.repository.get(new_aggregate.id)
    except AggregateNotFoundError:
        print("Aggregate not found (transaction rolled back)")
        # Output: Aggregate not found (transaction rolled back)
```

--------------------------------

### Handle Application Events with SQLAlchemyApplicationRecorder

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Extends aggregate recording with notification support, providing auto-incrementing notification IDs for event-driven architectures. Subscribers can poll for new events across all aggregates using this recorder.

```python
from eventsourcing.persistence import StoredEvent
from eventsourcing_sqlalchemy.datastore import SQLAlchemyDatastore
from eventsourcing_sqlalchemy.recorders import SQLAlchemyApplicationRecorder
from uuid import uuid4

# Create datastore and recorder
datastore = SQLAlchemyDatastore(url='sqlite:///:memory:')
recorder = SQLAlchemyApplicationRecorder(
    datastore=datastore,
    events_table_name='app_events',
    schema_name=None
)
recorder.create_table()

# Insert events (returns notification IDs)
originator_id = uuid4()
events = [
    StoredEvent(
        originator_id=originator_id,
        originator_version=1,
        topic='user:Created',
        state=b'{"email": "user@example.com"}'
    ),
    StoredEvent(
        originator_id=originator_id,
        originator_version=2,
        topic='user:EmailVerified',
        state=b'{}'
    ),
]
notification_ids = recorder.insert_events(events)

# Get max notification ID
max_id = recorder.max_notification_id()
print(f"Max notification ID: {max_id}")
# Output: Max notification ID: 2

# Select notifications for event polling
notifications = recorder.select_notifications(start=1, limit=10)
print(f"Notifications count: {len(notifications)}")
# Output: Notifications count: 2

```

--------------------------------

### Configure SQLAlchemy Scoped Sessions - Python

Source: https://github.com/pyeventsourcing/eventsourcing-sqlalchemy/blob/main/README.md

Shows how to configure a pyeventsourcing application to use an SQLAlchemy scoped_session. This involves defining an adapter for the scoped_session and setting the SQLALCHEMY_SCOPED_SESSION_TOPIC environment variable. It also demonstrates the necessary commit and remove calls for managing sessions within request lifecycles.

```python
from eventsourcing.application import AggregateNotFoundError
from eventsourcing.utils import get_topic
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

# Create engine.
engine = create_engine('sqlite:///:memory:')

# Create a scoped_session object.
session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)

# Define an adapter for the scoped session.
class MyScopedSessionAdapter:
    def __getattribute__(self, item: str) -> None:
        return getattr(session, item)

# Produce the topic of the scoped session adapter class.
scoped_session_topic = get_topic(MyScopedSessionAdapter)

# Construct an event-sourced application.
app = Application(
    env={'SQLALCHEMY_SCOPED_SESSION_TOPIC': scoped_session_topic}
)

# During request.
aggregate = Aggregate()
app.save(aggregate)
app.repository.get(aggregate.id)
session.commit()

# After request.
session.remove()

# During request.
app.repository.get(aggregate.id)

# After request.
session.remove()

# During request.
aggregate = Aggregate()
app.save(aggregate)
# forget to commit

# After request.
session.remove()

# During request.
try:
    # forgot to commit
    app.repository.get(aggregate.id)
except AggregateNotFoundError:
    pass
else:
    raise Exception("Expected aggregate not found")

# After request.
session.remove()
```

--------------------------------

### Configure Google Cloud SQL Connection

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Sets up a secure, IAM-authenticated connection to a Google Cloud SQL instance using the Cloud SQL Python Connector. This involves initializing the connector, defining a connection creator function, and configuring environment variables for the eventsourcing application.

```python
import os
from eventsourcing.application import Application
from eventsourcing.utils import get_topic

# Install: pip install 'cloud-sql-python-connector[pg8000]'
from google.cloud.sql.connector import Connector

# Initialize Cloud SQL connector
connector = Connector()

# Define connection creator function
def get_google_cloud_sql_conn():
    return connector.connect(
        "project:region:instance",
        "pg8000",
        user="postgres-iam-user@gmail.com",
        db="my-db-name",
        enable_iam_auth=True,
    )

# Configure environment
os.environ['PERSISTENCE_MODULE'] = 'eventsourcing_sqlalchemy'
os.environ['SQLALCHEMY_URL'] = 'postgresql+pg8000://'
os.environ['SQLALCHEMY_CONNECTION_CREATOR_TOPIC'] = get_topic(get_google_cloud_sql_conn)

# Create application with Cloud SQL
app = Application()
print("Connected to Google Cloud SQL")
```

--------------------------------

### Manage Transactions with SQLAlchemy Recorder

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Demonstrates using the recorder's transaction() method for context-managed transactions with ORM models. Supports nested transactions and automatic rollback on exceptions. Autoflush behavior can be controlled via environment variables or session maker configurations.

```python
import os
from eventsourcing.application import Application
from eventsourcing.domain import Aggregate

os.environ['PERSISTENCE_MODULE'] = 'eventsourcing_sqlalchemy'
os.environ['SQLALCHEMY_URL'] = 'sqlite:///:memory:'

app = Application()

# Manage transactions outside the application
with app.recorder.transaction(commit=True) as session:
    # You can add SQLAlchemy ORM objects to the session
    # session.add(my_orm_object)

    # And also save event-sourced aggregates
    aggregate = Aggregate()
    app.save(aggregate)

    # Both will be committed atomically
    print(f"Session autoflush: {session.autoflush}")
    # Output: Session autoflush: True

# Disable autoflush via environment
app2 = Application(env={'SQLALCHEMY_AUTOFLUSH': 'False'})
with app2.recorder.transaction(commit=True) as session:
    print(f"Session autoflush: {session.autoflush}")
    # Output: Session autoflush: False

# Disable autoflush via session maker
app.factory.datastore.session_maker.kw["autoflush"] = False
with app.recorder.transaction(commit=True) as session:
    print(f"Session autoflush: {session.autoflush}")
    # Output: Session autoflush: False

# Use no_autoflush context manager for temporary disable
with app.recorder.transaction(commit=True) as session:
    with session.no_autoflush:
        print(f"Session autoflush inside context: {session.autoflush}")
        # Output: Session autoflush inside context: False

```

--------------------------------

### Filter Notifications by Topic with SQLAlchemyProcessRecorder

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Demonstrates how to filter recorded notifications by specifying topics. This is useful for selecting specific types of events from the recorder.

```python
from eventsourcing_sqlalchemy.recorders import SQLAlchemyProcessRecorder
from eventsourcing_sqlalchemy.datastore import SQLAlchemyDatastore

# Assuming 'recorder' is an instance of SQLAlchemyProcessRecorder
datastore = SQLAlchemyDatastore(url='sqlite:///:memory:')
recorder = SQLAlchemyProcessRecorder(
    datastore=datastore,
    events_table_name='process_events',
    tracking_table_name='notification_tracking',
    schema_name=None
)
recorder.create_table()

notifications = recorder.select_notifications(
    start=1,
    limit=10,
    topics=['user:Created']
)
print(f"Filtered notifications: {len(notifications)}")
```

--------------------------------

### Store and Query Aggregate Events with SQLAlchemyAggregateRecorder

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Utilizes SQLAlchemyAggregateRecorder to store and retrieve domain events for aggregates. It dynamically creates tables and supports filtering events by originator ID and version range. Events are stored as StoredEvent objects.

```python
from eventsourcing.persistence import StoredEvent
from eventsourcing_sqlalchemy.datastore import SQLAlchemyDatastore
from eventsourcing_sqlalchemy.recorders import SQLAlchemyAggregateRecorder
from uuid import uuid4

# Create datastore and recorder
datastore = SQLAlchemyDatastore(url='sqlite:///:memory:')
recorder = SQLAlchemyAggregateRecorder(
    datastore=datastore,
    events_table_name='dog_events',
    schema_name=None,
    for_snapshots=False
)
recorder.create_table()

# Insert events
originator_id = uuid4()
events = [
    StoredEvent(
        originator_id=originator_id,
        originator_version=1,
        topic='dog:Registered',
        state=b'{"name": "Fido"}'
    ),
    StoredEvent(
        originator_id=originator_id,
        originator_version=2,
        topic='dog:TrickAdded',
        state=b'{"trick": "roll over"}'
    ),
]
recorder.insert_events(events)

# Select events
stored = recorder.select_events(originator_id=originator_id)
print(f"Events count: {len(stored)}")
# Output: Events count: 2

# Select with version filter
stored = recorder.select_events(originator_id=originator_id, gt=1)
print(f"Events after version 1: {len(stored)}")
# Output: Events after version 1: 1

# Select in descending order with limit
stored = recorder.select_events(originator_id=originator_id, desc=True, limit=1)
print(f"Latest event topic: {stored[0].topic}")
# Output: Latest event topic: dog:TrickAdded

```

--------------------------------

### FastAPI Scoped Session Adapter for Eventsourcing

Source: https://github.com/pyeventsourcing/eventsourcing-sqlalchemy/blob/main/README.md

Implements a session adapter for FastAPI applications using the fastapi-sqlalchemy library. It leverages the global `db.session` provided by the library, which offers request-scoped sessions. This adapter enables event-sourced applications to work with FastAPI's automatic session management, including commit-on-exit behavior.

```python
from fastapi import FastAPI
from fastapi_sqlalchemy import db, DBSessionMiddleware
from eventsourcing.application import Application
from eventsourcing.domain import Aggregate
from eventsourcing.utils import get_topic
from eventsourcing.exceptions import AggregateNotFoundError
import os

fastapi_app = FastAPI()
fastapi_app.add_middleware(
    DBSessionMiddleware, db_url='sqlite:///:memory:'
)
fastapi_app.build_middleware_stack()

class FastapiScopedSession:
    def __getattribute__(self, item: str) -> None:
        return getattr(db.session, item)

with db(commit_on_exit=True):
    scoped_session_adapter_topic = get_topic(FlaskScopedSession) # Note: This seems to be a typo in the original, should likely be FastapiScopedSession
    es_app = Application(
        env={"SQLALCHEMY_SCOPED_SESSION_TOPIC": get_topic(FastapiScopedSession)}
    )

with db(commit_on_exit=True):
    aggregate = Aggregate()
    es_app.save(aggregate)
    es_app.repository.get(aggregate.id)

with db(commit_on_exit=True):
    es_app.repository.get(aggregate.id)

try:
    with db(commit_on_exit=True):
        aggregate = Aggregate()
        es_app.save(aggregate)
        es_app.repository.get(aggregate.id)
        raise TypeError("An error occurred!!!")
except TypeError:
    pass
else:
    raise Exception("Expected type error")

with db(commit_on_exit=True):
    try:
        es_app.repository.get(aggregate.id)
    except AggregateNotFoundError:
        pass
    else:
        raise Exception("Expected aggregate not found")

```

--------------------------------

### Flask Scoped Session Adapter for Eventsourcing

Source: https://github.com/pyeventsourcing/eventsourcing-sqlalchemy/blob/main/README.md

Defines a Flask-scoped session adapter to be used with the eventsourcing library. It mimics SQLAlchemy's scoped_session behavior by delegating attribute access to `db.session`. This allows event-sourced applications to seamlessly use Flask's request-scoped database sessions.

```python
from eventsourcing.application import Application
from eventsourcing.domain import Aggregate
from eventsourcing.utils import get_topic

# Assume 'db' is a SQLAlchemy instance with a session attribute
# Assume 'flask_app' is a Flask application instance

class FlaskScopedSession:
    def __getattribute__(self, item: str) -> None:
        return getattr(db.session, item)

with flask_app.app_context():
    scoped_session_adapter_topic = get_topic(FlaskScopedSession)
    es_app = Application(
        env={"SQLALCHEMY_SCOPED_SESSION_TOPIC": scoped_session_adapter_topic}
    )

    aggregate = Aggregate()
    es_app.save(aggregate)
    db.session.commit()

    db.session.remove()

    es_app.repository.get(aggregate.id)

    db.session.remove()
```

--------------------------------

### Scoped Sessions with FastAPI-SQLAlchemy

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Integrates eventsourcing-sqlalchemy with FastAPI and fastapi-sqlalchemy middleware for automatic session management. Sessions are committed on success and rolled back on exceptions.

```python
from fastapi import FastAPI
from fastapi_sqlalchemy import db, DBSessionMiddleware
from eventsourcing.application import Application, AggregateNotFoundError
from eventsourcing.domain import Aggregate
from eventsourcing.utils import get_topic

fastapi_app = FastAPI()
fastapi_app.add_middleware(
    DBSessionMiddleware,
    db_url='sqlite:///:memory:'
)
fastapi_app.build_middleware_stack()

class FastapiScopedSession:
    def __getattribute__(self, item: str):
        return getattr(db.session, item)

with db(commit_on_exit=True):
    es_app = Application(
        env={"SQLALCHEMY_SCOPED_SESSION_TOPIC": get_topic(FastapiScopedSession)}
    )

with db(commit_on_exit=True):
    aggregate = Aggregate()
    es_app.save(aggregate)
    es_app.repository.get(aggregate.id)
```

--------------------------------

### Check Tracking ID Existence with SQLAlchemyProcessRecorder

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Demonstrates how to check if a specific notification ID has already been tracked for an application. This prevents duplicate processing of events.

```python
from eventsourcing.persistence import StoredEvent, Tracking
from eventsourcing_sqlalchemy.datastore import SQLAlchemyDatastore
from eventsourcing_sqlalchemy.recorders import SQLAlchemyProcessRecorder
from uuid import uuid4

datastore = SQLAlchemyDatastore(url='sqlite:///:memory:')
recorder = SQLAlchemyProcessRecorder(
    datastore=datastore,
    events_table_name='process_events',
    tracking_table_name='notification_tracking',
    schema_name=None
)
recorder.create_table()

originator_id = uuid4()
events = [
    StoredEvent(
        originator_id=originator_id,
        originator_version=1,
        topic='order:Created',
        state=b'{"order_id": "12345"}'
    ),
]
tracking = Tracking(application_name='upstream_app', notification_id=5)
recorder.insert_events(events, tracking=tracking)

has_tracking = recorder.has_tracking_id('upstream_app', 5)
print(f"Has tracking ID 5: {has_tracking}")

has_tracking = recorder.has_tracking_id('upstream_app', 6)
print(f"Has tracking ID 6: {has_tracking}")
```

--------------------------------

### Scoped Sessions with Flask-SQLAlchemy

Source: https://context7.com/pyeventsourcing/eventsourcing-sqlalchemy/llms.txt

Integrates eventsourcing-sqlalchemy with Flask-SQLAlchemy for request-scoped transactions. This ensures that sessions are automatically committed or rolled back correctly.

```python
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from eventsourcing.application import Application
from eventsourcing.domain import Aggregate
from eventsourcing.utils import get_topic

try:
    from sqlalchemy.orm import declarative_base
except ImportError:
    from sqlalchemy.ext.declarative import declarative_base

flask_app = Flask(__name__)
flask_app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'

Base = declarative_base()
db = SQLAlchemy(flask_app, model_class=Base)

class FlaskScopedSession:
    def __getattribute__(self, item: str):
        return getattr(db.session, item)

with flask_app.app_context():
    scoped_session_topic = get_topic(FlaskScopedSession)
    es_app = Application(
        env={"SQLALCHEMY_SCOPED_SESSION_TOPIC": scoped_session_topic}
    )

    aggregate = Aggregate()
    es_app.save(aggregate)
    db.session.commit()

    db.session.remove()

    retrieved = es_app.repository.get(aggregate.id)
    print(f"Retrieved aggregate ID: {retrieved.id}")

    db.session.remove()
```

--------------------------------

### Manage SQLAlchemy Transactions Outside Application - Python

Source: https://github.com/pyeventsourcing/eventsourcing-sqlalchemy/blob/main/README.md

Demonstrates how to manage SQLAlchemy ORM model updates atomically with event-sourced application updates using the application recorder's transaction() method. The returned Transaction object acts as a context manager to obtain an SQLAlchemy Session, allowing atomic commits when the context manager exits. This effectively implements thread-scoped transactions.

```python
with school.recorder.transaction() as session:
    # Update CRUD model.
    ...  # session.add(my_orm_object)
    # Update event-sourced application.
    school.register('Buster')
    school.add_trick('Buster', 'fetch ball')

    tricks = school.get_tricks('Buster')
    assert tricks == ['fetch ball']
```

```python
app = Application()

with app.recorder.transaction() as session:
    assert session.autoflush is True
```

```python
app = Application(env={'SQLALCHEMY_AUTOFLUSH': 'False'})

with app.recorder.transaction() as session:
    assert session.autoflush is False
```

```python
app = Application()
app.recorder.datastore.session_maker.kw["autoflush"] = False

with app.recorder.transaction() as session:
    assert session.autoflush is False
```

```python
app = Application()

with app.recorder.transaction() as session:
    with session.no_autoflush:
        assert session.autoflush is False
```

```python
app = Application()

with app.recorder.transaction() as session:
    session.autoflush = False
    # Add CRUD objects to the session.
    ...
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.