### Set up PostgreSQL database and user

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Commands to install PostgreSQL, start the service, create the 'eventsourcing' database and user, and set ownership. This is required for tests that interact with PostgreSQL.

```bash
$ brew install postgresql
$ brew services start postgresql
$ psql postgres
postgres=# CREATE DATABASE eventsourcing;
postgres=# CREATE USER eventsourcing WITH PASSWORD 'eventsourcing';
postgres=# ALTER DATABASE eventsourcing OWNER TO eventsourcing;
$ psql eventsourcing
postgres=# CREATE SCHEMA myschema AUTHORIZATION eventsourcing;
```

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

### Constructing and Using Wiki Application

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/application.rst

Instantiates the Wiki application and demonstrates creating a new page. This example shows the initial setup and usage of the application.

```python
wiki = Wiki()
```

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

### Install eventsourcing with optional dependencies

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Install the library with specific optional dependencies like postgres and cryptography.

```bash
$ pip install "eventsourcing[postgres,cryptography]"
```

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

### Install development dependencies

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Creates a virtual environment and installs all necessary packages for developing the library, including testing and linting tools.

```bash
$ make install
```

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

### Install Poetry for development

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Installs the required version of Poetry using pipx. This is a prerequisite for other development setup steps.

```bash
$ make install-poetry
```

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

### Install eventsourcing with pip

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Install the library using pip. It is recommended to do this within a Python virtual environment.

```bash
$ pip install eventsourcing
```

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

### CourseSubscriptions Application Setup

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/dcb.rst

Constructs a CourseSubscriptions application using MessagePack for serialization and in-memory persistence. This setup is useful for testing or simple applications.

```python
app = CourseSubscriptions(env={
        "PERSISTENCE_MODULE": "eventsourcing.dcb.popo",
        "MAPPER_TOPIC": "eventsourcing.dcb.msgpack:MessagePackMapper",
    })
```

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

### POPOApplicationRecorder Example

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Demonstrates constructing and using POPOApplicationRecorder to insert and select stored events.

```python
from eventsourcing.popo import POPOApplicationRecorder

# Construct application recorder.
application_recorder = POPOApplicationRecorder()

# Insert stored events.
application_recorder.insert_events([stored_event])

# Select stored events from an aggregate sequence.
recorded_events = application_recorder.select_events(stored_event.originator_id)
assert recorded_events[0] == stored_event
```

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

### Install eventsourcing with version pinning

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Install a specific major and minor version of the library to avoid breaking changes. This is useful for application development.

```bash
$ pip install "eventsourcing~=9.5.5"
```

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

### PostgreSQL Application Recorder Example

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Shows how to create a table, insert events, select events by originator ID, and select notifications using PostgresApplicationRecorder.

```python
from eventsourcing.postgres import PostgresApplicationRecorder

# Construct application recorder and create table.
application_recorder = PostgresApplicationRecorder(datastore)
application_recorder.create_table()

# Insert stored events.
application_recorder.insert_events([stored_event])

# Select stored events from an aggregate sequence.
recorded_events = application_recorder.select_events(stored_event.originator_id)
assert recorded_events[0] == stored_event

# Select notifications from the application sequence.
notifications = application_recorder.select_notifications(start=1, limit=10)
assert notifications[0].id == 1
assert notifications[0].originator_id == stored_event.originator_id
assert notifications[0].originator_version == stored_event.originator_version
```

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

### Constructing EventStore with Factory

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Constructs an EventStore using the factory, providing the mapper and recorder. Includes example of putting and getting events.

```python
event_store = factory.event_store(
    mapper=mapper,
    recorder=recorder,
)

event_store.put([domain_event])
domain_events = list(event_store.get(id1))

assert domain_events == [domain_event]
```

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

### PostgreSQL Aggregate Recorder Example

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Demonstrates creating a table and inserting/selecting stored events using PostgresAggregateRecorder.

```python
from eventsourcing.postgres import PostgresAggregateRecorder

# Construct aggregate recorder and create table.
aggregate_recorder = PostgresAggregateRecorder(datastore)
aggregate_recorder.create_table()

# Insert stored events.
aggregate_recorder.insert_events([stored_event])

# Select stored events from an aggregate sequence.
recorded_events = aggregate_recorder.select_events(stored_event.originator_id)
assert recorded_events[0] == stored_event, (recorded_events[0], stored_event)
```

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

### Install PostgreSQL with Homebrew

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/dev/MACOS_SETUP_NOTES.md

Use Homebrew to install PostgreSQL on your macOS system. This is the initial step for setting up the database.

```bash
brew install postgresql
```

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

### Install PostgreSQL and Crypto Dependencies

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/tutorial/part3.rst

Commands to install the eventsourcing library with PostgreSQL and encryption support. Install either psycopg[c,pool] or psycopg[binary,pool] for PostgreSQL, and pycryptodome for encryption.

```bash
$ pip install eventsourcing[postgres]
```

```bash
$ pip install eventsourcing[crypto]
```

```bash
$ pip install eventsourcing[crypto,postgres]
```

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

### SQLite Tracking Recorder Example

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Demonstrates how to use SQLiteTrackingRecorder to create a table, insert tracking objects, and query tracking information.

```python
from eventsourcing.sqlite import SQLiteTrackingRecorder

# Construct tracking recorder and create table.
tracking_recorder = SQLiteTrackingRecorder(datastore)
tracking_recorder.create_table()

# Construct tracking object.
tracking = Tracking(notification_id=21, application_name="upstream")

# Insert tracking object.
tracking_recorder.insert_tracking(tracking=tracking)

# Get latest tracked position.
assert tracking_recorder.max_tracking_id("upstream") == 21

# Check if an event notification has been processed.
assert tracking_recorder.has_tracking_id("upstream", 21)
assert not tracking_recorder.has_tracking_id("upstream", 22)
```

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

### Run the test suite

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Executes the project's test suite to ensure the library is functioning correctly after installation.

```bash
$ make test
```

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

### Run all development tasks

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Executes a comprehensive set of development tasks including installation, documentation build, linting, and testing.

```bash
$ make install docs fmt lint test benchmark
```

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

### Install eventsourcing with PostgreSQL support

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Install the library with the 'postgres' extra to include dependencies for storing events with PostgreSQL, such as Psycopg v3.

```bash
$ pip install "eventsourcing[postgres]"
```

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

### POPOAggregateRecorder Example

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Demonstrates constructing and using POPOAggregateRecorder to insert and select stored events.

```python
from eventsourcing.popo import POPOAggregateRecorder

# Construct aggregate recorder.
aggregate_recorder = POPOAggregateRecorder()

# Insert stored events.
aggregate_recorder.insert_events([stored_event])

# Select stored events from an aggregate sequence.
recorded_events = aggregate_recorder.select_events(stored_event.originator_id)
assert recorded_events[0] == stored_event
```

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

### Install eventsourcing with PyCryptodome support

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Install the library with the 'crypto' extra to include PyCryptodome for storing cryptographically encrypted events.

```bash
$ pip install "eventsourcing[crypto]"
```

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

### SQLite Process Recorder Example

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Shows how to use SQLiteProcessRecorder to create a table, insert events atomically with tracking objects, and select recorded events and notifications.

```python
from eventsourcing.sqlite import SQLiteProcessRecorder

# Create an SQLite database.
datastore = SQLiteDatastore(db_name=":memory:")

# Construct process recorder and create database table.
process_recorder = SQLiteProcessRecorder(datastore)
process_recorder.create_table()

# Construct a tracking object.
tracking = Tracking(notification_id=21, application_name="upstream")

# Insert stored events atomically with a tracking object.
process_recorder.insert_events([stored_event], tracking=tracking)

# Select stored events from an aggregate sequence.
recorded_events = process_recorder.select_events(stored_event.originator_id)

# Select notifications from the application sequence.
notifications = process_recorder.select_notifications(start=1, limit=10)

# Get latest tracked position.
assert process_recorder.max_tracking_id("upstream") == 21
```

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

### Start PostgreSQL Service

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/dev/MACOS_SETUP_NOTES.md

Start the PostgreSQL service using Homebrew services. This command ensures PostgreSQL runs in the background.

```bash
brew services start postgresql
```

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

### Install libpq for Psycopg

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/dev/MACOS_SETUP_NOTES.md

Install the libpq library using Homebrew. This is required if you need to use the psycopg library without its binary distributions, especially when testing new Python releases.

```bash
brew install libpq
```

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

### Install eventsourcing with cryptography support

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Install the library with the 'cryptography' extra to include the Python cryptography package for storing encrypted events.

```bash
$ pip install "eventsourcing[cryptography]"
```

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

### Run Projection with DCB Application Class

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/projection.rst

Shows how to run a projection with a DCB (Domain Command Bus) application class. Similar to the standard application example, it uses the ProjectionRunner as a context manager and runs indefinitely. The example includes a thread to simulate an interrupt signal after 1 second and registers a signal handler for graceful shutdown. Environment variables for the mapper topic are also configured.

```python
import os, signal, threading, time

from eventsourcing.projection import ProjectionRunner

# For demonstration purposes, interrupt process with SIGINT after 1s.
def sleep_then_kill() -> None:
    time.sleep(1)
    os.kill(os.getpid(), signal.SIGINT)

threading.Thread(target=sleep_then_kill).start()

# Run projection as a context manager.
with ProjectionRunner(
    application_class=DCBApplication,
    view_class=MyPOPOMaterialisedView,
    projection_class=TaggedDecisionProjection,
    env={\"MAPPER_TOPIC\": get_topic(MessagePackMapper)},
) as projection_runner:

    # Register signal handler.
    signal.signal(signal.SIGINT, lambda *args: projection_runner.stop())

    # Run until interrupted.
    projection_runner.run_forever()

```

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

### Create and activate a virtual environment

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Create a Python virtual environment and activate it before installing the library. This isolates project dependencies.

```bash
python3 -m venv my_venv
source my_venv/bin/activate
(my_venv) $ pip install eventsourcing
```

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

### SQLiteApplicationRecorder Usage

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Illustrates the setup and usage of SQLiteApplicationRecorder for creating tables, inserting events, selecting events by aggregate ID, and selecting notifications from the application sequence.

```python
from eventsourcing.sqlite import SQLiteApplicationRecorder

# Create an SQLite database.
datastore = SQLiteDatastore(db_name=":memory:")

# Construct application recorder and create database table.
application_recorder = SQLiteApplicationRecorder(datastore)
application_recorder.create_table()

# Insert stored events.
application_recorder.insert_events([stored_event])

# Select stored events from an aggregate sequence.
recorded_events = application_recorder.select_events(stored_event.originator_id)

# Select notifications from the application sequence.
notifications = application_recorder.select_notifications(start=1, limit=10)
```

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

### Write and Run a Test for DogSchool

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/README.md

Write a test function to verify the functionality of the `DogSchool` application. This example uses in-memory persistence by default.

```python
def test_dog_school() -> None:
    # Construct application object.
    school = DogSchool()

    # Evolve application state.
    dog_id = school.register_dog('Fido')
    school.add_trick(dog_id, 'roll over')
    school.add_trick(dog_id, 'play dead')

    # Query application state.
    dog = school.get_dog(dog_id)
    assert dog['name'] == 'Fido'
    assert dog['tricks'] == ('roll over', 'play dead')

    # Select notifications.
    notifications = school.notification_log.select(start=1, limit=10)
    assert len(notifications) == 3
```

```python
test_dog_school()
```

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

### Set up Environment Variables for Eventsourcing

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

Configure persistence, cipher, and compressor modules by setting environment variables. This example uses an in-memory SQLite database and AES-256 encryption with Zlib compression.

```python
cipher_key = AESCipher.create_key(32)  # 32-byte key → AES-256

os.environ["PERSISTENCE_MODULE"] = "eventsourcing.sqlite"
os.environ["SQLITE_DBNAME"] = ":memory:"
os.environ["CIPHER_TOPIC"] = "eventsourcing.cipher:AESCipher"
os.environ["CIPHER_KEY"] = cipher_key
os.environ["COMPRESSOR_TOPIC"] = "eventsourcing.compressor:ZlibCompressor"
```

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

### Page and Index Aggregate Usage Example

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/domain.rst

Demonstrates creating, updating, and recreating Page and Index aggregates. Shows how to use aggregate IDs and names to link and retrieve related aggregates.

```python
# Create new page and index aggregates.
page = Page(name="Erth")
index1 = Index(name=page.name, ref=page.id)

# The page name can be used to recreate
# the index ID. The index ID can be used
# to retrieve the index aggregate, which
# gives the page ID, and then the page ID
# can be used to retrieve the page aggregate.
index_id = Index.create_id(name="Erth")
assert index_id == index1.id
assert index1.ref == page.id
assert index1.name == page.name

# Later, the page name can be updated,
# and a new index created for the page.
page.update_name(name="Earth")
index1.update_ref(ref=None)
index2 = Index(name=page.name, ref=page.id)

# The new page name can be used to recreate
# the new index ID. The new index ID can be
# used to retrieve the new index aggregate,
# which gives the page ID, and then the page
# ID can be used to retrieve the renamed page.
index_id = Index.create_id(name="Earth")
assert index_id == index2.id
assert index2.ref == page.id
assert index2.name == page.name
```

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

### Using EventStore to Put and Get Events

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Demonstrates basic usage of the EventStore for saving and retrieving domain events. Requires an initialized EventStore with a mapper and recorder.

```python
from eventsourcing.persistence import EventStore

application_recorder = POPOApplicationRecorder()

event_store = EventStore[
    UUID
](
    mapper=mapper,
    recorder=application_recorder,
)

event_store.put([domain_event])

domain_events = list(event_store.get(id1))
assert domain_events == [domain_event]
```

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

### Create PostgreSQL Roles and Databases

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/dev/MACOS_SETUP_NOTES.md

Connect to PostgreSQL and create necessary roles ('postgres', 'eventsourcing') and databases ('eventsourcing', 'eventsourcing_nopublic'). This setup is required for the 'eventsourcing' project.

```bash
psql postgres
```

```sql
CREATE ROLE postgres LOGIN SUPERUSER PASSWORD 'postgres';
CREATE USER eventsourcing WITH PASSWORD 'eventsourcing';
CREATE DATABASE eventsourcing;
ALTER DATABASE eventsourcing OWNER TO eventsourcing;
CREATE DATABASE eventsourcing_nopublic;
ALTER DATABASE eventsourcing_nopublic OWNER TO eventsourcing;
```

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

### Construct and Use SQLite Infrastructure

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Constructs the infrastructure factory for SQLite and demonstrates basic event store operations like putting and getting events.

```python
factory = InfrastructureFactory.construct(environ)
recorder = factory.application_recorder()
assert isinstance(recorder, SQLiteApplicationRecorder)

mapper = factory.mapper(transcoder=transcoder)
event_store = factory.event_store(
    mapper=mapper,
    recorder=recorder,
)

event_store.put([domain_event])
domain_events = list(event_store.get(id1))

assert domain_events == [domain_event]
```

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

### Construct and Use PostgreSQL Infrastructure

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Constructs the infrastructure factory for PostgreSQL, creates an application recorder and event store, and demonstrates putting and getting domain events. Ensure the environment is configured for PostgreSQL.

```python
factory = InfrastructureFactory.construct(environ)
recorder = factory.application_recorder()
assert isinstance(recorder, PostgresApplicationRecorder)

mapper = factory.mapper(transcoder=transcoder)
event_store = factory.event_store(
    mapper=mapper,
    recorder=recorder,
)

event_store.put([domain_event])
domain_events = list(event_store.get(id1))

assert domain_events == [domain_event]
```

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

### Write Test for Dog School Application

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/tutorial/part1.rst

Example of how to write a unit test for the DogSchool application. It covers registering a dog, adding tricks, and verifying the state changes.

```python
def test_dog_school() -> None:

        # Construct the application.
        app = DogSchool()

        # Register a dog.
        dog_id = app.register_dog(name='Fido')

        # Check the dog has been registered.
        assert app.get_dog(dog_id) == {
            'name': 'Fido',
            'tricks': (),
        }

        # Add tricks.
        app.add_trick(dog_id, trick='roll over')
        app.add_trick(dog_id, trick='fetch ball')

        # Check the tricks have been added.
        assert app.get_dog(dog_id) == {
            'name': 'Fido',
            'tricks': ('roll over', 'fetch ball'),
        }
```

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

### Instantiate and use DogSchool application

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/tutorial/part3.rst

Demonstrates constructing a DogSchool application instance and calling its command and query methods to register a dog, add tricks, and retrieve dog details. Asserts the correctness of the retrieved dog information.

```python
application = DogSchool()

dog_id = application.register_dog(name='Fido')
application.add_trick(dog_id, trick='roll over')
application.add_trick(dog_id, trick='fetch ball')

dog_details = application.get_dog(dog_id)
assert dog_details['name'] == 'Fido'
assert dog_details['tricks'] == ('roll over', 'fetch ball')
```

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

### Use DogSchool Application

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/application.rst

Demonstrates creating a DogSchool application instance, registering a dog, adding multiple tricks, and retrieving the list of tricks.

```python
dog_school = DogSchool()

dog_id = dog_school.register_dog()

dog_school.add_trick(dog_id, "roll over")
dog_school.add_trick(dog_id, "fetch ball")
dog_school.add_trick(dog_id, "play dead")

tricks = dog_school.get_tricks(dog_id)
assert tricks[0] == "roll over"
assert tricks[1] == "fetch ball"
assert tricks[2] == "play dead"
```

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

### Instantiate DCBRepository

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/dcb.rst

Construct a DCBRepository with an event store. Requires MessagePackMapper and InMemoryDCBRecorder.

```python
from eventsourcing.dcb.application import DCBRepository

repository = DCBRepository(
    eventstore=DCBEventStore(
        mapper=MessagePackMapper(),
        recorder=InMemoryDCBRecorder(),
    ),
)
```

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

### Example Usage of System Module

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/system.rst

This snippet demonstrates a typical workflow involving registering dogs, adding tricks, and asserting counts. It includes a small delay to ensure asynchronous operations complete before assertions.

```python
dog_id2 = school.register_dog('Milly')
dog_id3 = school.register_dog('Scrappy')

school.add_trick(dog_id1, 'roll over')
school.add_trick(dog_id2, 'roll over')
school.add_trick(dog_id3, 'roll over')

school.add_trick(dog_id1, 'fetch ball')
school.add_trick(dog_id2, 'fetch ball')

school.add_trick(dog_id1, 'play dead')

from time import sleep

sleep(0.01)

assert counters.get_count('roll over') == 3
assert counters.get_count('fetch ball') == 2
assert counters.get_count('play dead') == 1

runner.stop()
```

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

### DCBApplication Example: Student Enrollment

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

Demonstrates the Dynamic Consistency Boundary pattern using DCBApplication for managing student enrollments in courses. It defines custom Decision and Slice classes to handle enrollment logic, including checking for existing enrollments and course capacity.

```python
from eventsourcing.dcb.application import DCBApplication
from eventsourcing.dcb.domain import Decision, InitialDecision, Slice, Selector

class StudentEnrolled(InitialDecision):
    student_id: str
    course_id: str

class CourseCapacityReached(Decision):
    course_id: str

class EnrolmentSlice(Slice):
    do_projection = True

    def __init__(self, student_id: str, course_id: str, capacity: int) -> None:
        self.student_id = student_id
        self.course_id = course_id
        self.capacity = capacity
        self.is_enrolled = False
        self.current_enrolments = 0

    def selector(self) -> list[Selector]:
        return [
            Selector(types=["StudentEnrolled"], tags={"student_id": self.student_id}),
            Selector(types=["StudentEnrolled", "CourseCapacityReached"],
                     tags={"course_id": self.course_id}),
        ]

    def project(self, event: Decision) -> None:
        if isinstance(event, StudentEnrolled):
            if event.student_id == self.student_id:
                self.is_enrolled = True
            elif event.course_id == self.course_id:
                self.current_enrolments += 1

    def execute(self) -> None:
        if self.is_enrolled:
            raise ValueError("Student already enrolled")
        if self.current_enrolments >= self.capacity:
            raise ValueError("Course is full")
        self.decide(StudentEnrolled(
            student_id=self.student_id,
            course_id=self.course_id,
        ))

class EnrolmentApp(DCBApplication):
    def enrol(self, student_id: str, course_id: str, capacity: int = 30) -> None:
        self.do(EnrolmentSlice(student_id, course_id, capacity))

app = EnrolmentApp()
app.enrol("student-1", "course-A")
app.enrol("student-2", "course-A")
print("Both enrolled successfully")
```

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

### Get Cart Items Query

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/examples/shop-vertical.rst

Query to retrieve items currently in the shopping cart. This snippet is part of the 'Get cart items' vertical slice.

```python
class GetCartItems:
    def __init__(self, cart_id: CartId) -> None:
        self.cart_id = cart_id
```

```python
class CartItem:
    def __init__(self, product_id: ProductId, quantity: int, price: Decimal) -> None:
        self.product_id = product_id
        self.quantity = quantity
        self.price = price
```

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

### Initialize Mapper with Transcoder

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Demonstrates how to create a Mapper instance, which is essential for converting domain events to stored events and vice versa. It requires a transcoder.

```python
from eventsourcing.persistence import Mapper

mapper = Mapper[UUID](transcoder=transcoder)
```

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

### PostgreSQL Datastore Configuration

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Illustrates how to configure and construct a PostgresDatastore object for connecting to a PostgreSQL database.

```python
from eventsourcing.postgres import PostgresDatastore

# Construct datastore object.
datastore = PostgresDatastore(
    dbname = "eventsourcing",
    host = "127.0.0.1",
    port = "5432",
    user = "eventsourcing",
    password = "eventsourcing",
)
```

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

### Get and Resolve Topics for Aggregate and Events

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/domain.rst

Demonstrates how to use get_topic to obtain the topic string for a class and resolve_topic to get the class from its topic string. This is essential for serializing domain events and identifying aggregate classes.

```python
from eventsourcing.utils import get_topic, resolve_topic


assert get_topic(Aggregate) == "eventsourcing.domain:Aggregate"
assert resolve_topic("eventsourcing.domain:Aggregate") == Aggregate

assert get_topic(Aggregate.Created) == "eventsourcing.domain:Aggregate.Created"
assert resolve_topic("eventsourcing.domain:Aggregate.Created") == Aggregate.Created
```

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

### Select event notifications from log

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/application.rst

Use the `select` method with `start` and `limit` arguments to retrieve a sub-sequence of event notifications. The selection will contain no more than the specified `limit` and notification IDs will be greater than or equal to `start`. By default, the first notification ID is 1.

```python
notifications = dog_school.notification_log.select(start=1, limit=2)
```

```python
assert len(notifications) == 2
```

```python
from eventsourcing.persistence import Notification

assert isinstance(notifications[0], Notification)
assert isinstance(notifications[1], Notification)
```

```python
assert notifications[0].id == 1
assert notifications[1].id == 2
```

```python
notification = notifications[0]
assert "Dog.Created" in notification.topic
assert notification.originator_id == dog_id

notification = notifications[1]
assert "Dog.TrickAdded" in notification.topic
assert b"roll over" in notification.state
assert notification.originator_id == dog_id
```

```python
notifications = dog_school.notification_log.select(
    start=notifications[-1].id + 1, limit=2
)

assert notifications[0].id == 3
assert notifications[1].id == 4

notification = notifications[0]
assert "Dog.TrickAdded" in notification.topic
assert b"fetch ball" in notification.state

notification = notifications[1]
assert "Dog.TrickAdded" in notification.topic
assert b"play dead" in notification.state
```

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

### Application class

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/examples/aggregate8.rst

The DogSchool application uses the PydanticApplication class from example 7.

```python
from eventsourcing.application import Application
from eventsourcing.application.sqlalchemy import SQLAlchemySettings
from eventsourcing.application.sqlalchemy.pydantic import PydanticApplication

from .domainmodel import DogSchool


class DogSchoolApplication(PydanticApplication):
    def __init__(self, **kwargs):
        super().__init__(DogSchool, **kwargs)

    def add_dog(self, name: str, age: int):
        self.dog_school.add_dog(name=name, age=age)

    def train_dog(self, name: str):
        self.dog_school.train_dog(name=name)

    def decommission_dog(self, name: str):
        self.dog_school.decommission_dog(name=name)

    @property
    def dog_school(self) -> DogSchool:
        return self.get_aggregate(DogSchool)


class DogSchoolApplicationWithSQLAlchemy(DogSchoolApplication):
    def __init__(self, db_connection_uri: str, **kwargs):
        settings = SQLAlchemySettings(db_connection_uri=db_connection_uri)
        super().__init__(settings=settings, **kwargs)

```

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

### Build Event-Driven System with ProcessApplication

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

Define an event-driven system using ProcessApplication for reacting to domain events. This example shows an Orders application and a Payments application that confirms small orders automatically.

```python
from uuid import UUID
from eventsourcing.application import Application, ProcessingEvent
from eventsourcing.domain import Aggregate, event
from eventsourcing.system import System, ProcessApplication
from eventsourcing.dispatch import singledispatchmethod

class Order(Aggregate):
    @event("Placed")
    def __init__(self, customer_id: str, total: float) -> None:
        self.customer_id = customer_id
        self.total = total
        self.status = "placed"

    @event("Confirmed")
    def confirm(self) -> None:
        self.status = "confirmed"

class Orders(Application[UUID]):
    def place_order(self, customer_id: str, total: float) -> UUID:
        order = Order(customer_id=customer_id, total=total)
        self.save(order)
        return order.id

class Payments(ProcessApplication[UUID]):
    """Reacts to Order.Placed events and auto-confirms small orders."""

    @singledispatchmethod
    def policy(
        self,
        domain_event: object,
        processing_event: ProcessingEvent,
    ) -> None:
        pass  # ignore unrecognized events

    @policy.register(Order.Placed)
    def _(self, domain_event: Order.Placed, processing_event: ProcessingEvent) -> None:
        if domain_event.total < 1000:
            # Retrieve the order and confirm it
            order: Order = self.repository.get(domain_event.originator_id)
            order.confirm()
            processing_event.collect_events(order)

# Define the system topology
system = System(pipes=[[Orders, Payments]])

# Run the system in-process with the SingleThreadedRunner
from eventsourcing.system import SingleThreadedRunner

import os
os.environ["PERSISTENCE_MODULE"] = "eventsourcing.popo"

with SingleThreadedRunner(system) as runner:
    orders_app: Orders = runner.get(Orders)
    payments_app: Payments = runner.get(Payments)

    order_id = orders_app.place_order("cust-42", total=99.0)
    runner.tick()  # process pending notifications

    order = orders_app.repository.get(order_id)
    print(order.status)  # "confirmed"
```

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

### Instantiate DogSchool application

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/tutorial/part3.rst

Shows how to construct an instance of the DogSchool application class. Asserts that the created object is an instance of the base Application class.

```python
application = DogSchool()

assert isinstance(application, Application)
```

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

### Immutable aggregate definition with Pydantic

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/examples/aggregate7.rst

An example of defining an immutable aggregate in a functional style using Pydantic.

```python
from eventsourcing.domain import Aggregate, Event
from pydantic.v1 import dataclasses


@dataclasses.dataclass
class Trick(ImmutableEvent):
    name: str


class DogSchool(Aggregate):
    class Event(ImmutableEvent):
        pass

    class Command(object):
        pass

    def __init__(self, name: str, **kwargs):
        super().__init__(**kwargs)
        self.name = name
        self.tricks = []

    def add_trick(self, name: str):
        self.tricks.append(Trick(name=name))

    def apply(self, event: Event):
        if isinstance(event, DogSchool.Event.TricksAdded):
            self.tricks.extend(event.tricks)


```

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

### Build documentation

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Builds the project's documentation and checks that the build process completes successfully.

```bash
$ make docs
```

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

### Update locked dependencies

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

Updates the locked package dependencies in the project and installs them. This ensures all development dependencies are up-to-date.

```bash
$ make update
```

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

### Run all development tasks (simplified)

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/installing.rst

A simplified command to run all essential development tasks.

```bash
$ make all
```

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

### Get latest aggregate from repository

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/application.rst

Retrieves the latest version of an aggregate from the repository. Asserts the number of tricks and the version.

```python
dog_latest: Dog = dog_school.repository.get(dog_id)

assert len(dog_latest.tricks) == 3
assert dog_latest.version == 4
```

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

### Test Get Cart Items

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/examples/shop-vertical.rst

Unit test for the 'GetCartItems' query. Verifies that cart items are correctly retrieved.

```python
class TestGetCartItems:
    def test_get_cart_items(self) -> None:
        query = GetCartItems(cart_id=CartId())
        items = query.execute()
        assert len(items) == 1
        assert isinstance(items[0], CartItem)
        assert items[0].quantity == 1
```

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

### Configure Application Environment Variables

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/application.rst

Shows how to configure an application's environment using a combination of class attributes, operating system environment variables, and constructor arguments. Demonstrates the order of precedence for these settings.

```python
import os

# Configure by setting class attribute.
class MyApplication(Application[UUID]):
    env = {"SETTING_A": "1", "SETTING_B": "1", "SETTING_C": "1"}

# Configure by setting operating system environment.
os.environ["SETTING_B"] = "2"
os.environ["SETTING_C"] = "2"

# Configure by setting constructor argument.
app = MyApplication(env={"SETTING_C": "3"})

assert app.env.get("SETTING_A") == "1"
assert app.env.get("SETTING_B") == "2"
assert app.env.get("SETTING_C") == "3"
```

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

### Configure persistence and infrastructure via environment variables

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

Illustrates how to configure the eventsourcing library's persistence and infrastructure settings using environment variables. This allows switching between storage backends like in-memory POPO, SQLite, and PostgreSQL without code changes.

```python
import os

# --- In-memory POPO (default, no config needed) ---
os.environ["PERSISTENCE_MODULE"] = "eventsourcing.popo"

# --- SQLite ---
os.environ["PERSISTENCE_MODULE"] = "eventsourcing.sqlite"
os.environ["SQLITE_DBNAME"] = "/path/to/db.sqlite3"       # or ":memory:"
os.environ["SQLITE_POOL_SIZE"] = "5"                       # optional connection pool

# --- PostgreSQL (requires psycopg[pool]) ---
os.environ["PERSISTENCE_MODULE"] = "eventsourcing.postgres"
os.environ["POSTGRES_DBNAME"] = "mydb"
os.environ["POSTGRES_HOST"] = "localhost"
os.environ["POSTGRES_PORT"] = "5432"
os.environ["POSTGRES_USER"] = "postgres"
os.environ["POSTGRES_PASSWORD"] = "secret"

# --- Encryption ---
os.environ["CIPHER_TOPIC"] = "eventsourcing.cipher:AESCipher"
os.environ["CIPHER_KEY"] = "<base64-encoded-32-byte-key>"   # AES-256

# --- Compression ---
os.environ["COMPRESSOR_TOPIC"] = "eventsourcing.compressor:ZlibCompressor"

# --- Snapshotting ---
os.environ["IS_SNAPSHOTTING_ENABLED"] = "y"

# --- Aggregate cache ---
os.environ["AGGREGATE_CACHE_MAXSIZE"] = "100"   # LRU cache size
os.environ["AGGREGATE_CACHE_FASTFORWARD"] = "y" # fast-forward cached aggregates

# --- Application-level env dict (overrides os.environ per application) ---
from eventsourcing.application import Application

class MyApp(Application):
    env = {
        "PERSISTENCE_MODULE": "eventsourcing.sqlite",
        "SQLITE_DBNAME": ":memory:",
    }

app = MyApp()  # uses its own env, not os.environ
```

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

### Create and Retrieve Wiki Page (Python)

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/application.rst

Demonstrates creating a wiki page with a name and body, and then retrieving its body using the page name.

```python
assert wiki.get_page(name="Erth").body == "Lorem ipsum..."
```

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

### Configure SQLite Persistence

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/application.rst

Sets up the application to use the SQLite persistence module by configuring environment variables for the persistence module and database name. This example also shows registering a dog and adding tricks.

```python
from tempfile import NamedTemporaryFile

tmpfile = NamedTemporaryFile(suffix="_eventsourcing_test.db")

os.environ["PERSISTENCE_MODULE"] = "eventsourcing.sqlite"
os.environ["SQLITE_DBNAME"] = tmpfile.name
application = DogSchool()

dog_id = application.register_dog()

application.add_trick(dog_id, "roll over")
application.add_trick(dog_id, "fetch ball")
application.add_trick(dog_id, "play dead")
```

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

### Get flat list of notifications as JSON

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

Retrieves a list of notifications and prints their type, which is expected to be a string representing JSON.

```python
json_notifications = service.get_notifications(start=1, limit=10)
print(type(json_notifications))  # str (JSON)
```

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

### Define DCBApplication Subclass

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/dcb.rst

Example of defining a DCBApplication subclass to manage domain objects and persistence. Includes a register_student method.

```python
from eventsourcing.dcb.application import DCBApplication


class CourseSubscriptions(DCBApplication):
    def register_student(self, name: str) -> str:
        student = Student(name=name, max_courses=5)
```

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

### Initialize and Use DCBEventStore

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/dcb.rst

Initialize a DCBEventStore with a mapper and recorder. Use read to retrieve events and append to add new tagged decisions.

```python
from eventsourcing.dcb.persistence import DCBEventStore

event_store = DCBEventStore(
    mapper=json_mapper,
    recorder=in_memory_recorder,
)

# Read already appended student events.
student_events = event_store.read(
    cb=student_selector,
)

# Append new course events.
event_store.append(
    events=[tagged_decision],
    cb=course_selector,
)

# Read already appended course events.
course_events = event_store.read(
    cb=course_selector,
)
```

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

### Get Dog Details

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/tutorial/part1.rst

Retrieve the current state of a dog aggregate using its ID. This method reconstructs the aggregate from its stored events.

```python
dog_details = application.get_dog(dog_id)
```

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

### Initialize Mapper with Compressor

Source: https://github.com/pyeventsourcing/eventsourcing/blob/9.5/docs/topics/persistence.rst

Demonstrates initializing a Mapper with a ZlibCompressor to compress and decompress stored event states. This can reduce storage size and transport time.

```python
from eventsourcing.compressor import ZlibCompressor

compressor = ZlibCompressor()

mapper = Mapper(
    transcoder=transcoder,
    compressor=compressor,
)

compressed_stored_event = mapper.to_stored_event(domain_event)
assert mapper.to_domain_event(compressed_stored_event) == domain_event
```