### Set up PostgreSQL Database and User

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

Installs PostgreSQL on macOS, starts the service, creates a database named 'eventsourcing', and a user with password 'eventsourcing'. It also sets the database owner and creates a schema.

```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;

```

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

### Install eventsourcing

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

Use this command to install the library from the Python Package Index.

```bash
$ pip install eventsourcing
```

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

### Install Development Dependencies

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

Creates a virtual environment and installs necessary packages for development, such as linters, formatters, and testing tools.

```bash
$ make install

```

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

### In-memory DCB Recorder Example

Source: https://eventsourcing.readthedocs.io/en/latest/topics/dcb.html

Example of using the `InMemoryDCBRecorder` for appending and reading events.

```APIDOC
## InMemoryDCBRecorder

### Description
Implements the `DCBRecorder` interface using only Python objects for in-memory storage.

### Usage Example

```python
from eventsourcing.dcb.popo import InMemoryDCBRecorder

in_memory_recorder = InMemoryDCBRecorder()

# Conditionally append new events.
in_memory_recorder.append(
    events=[student_registered],
    condition=append_condition,
)

# Read previously appended events.
student_events = in_memory_recorder.read(
    query=student_query,
)
```
```

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

### Install eventsourcing in a virtual environment

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

It is recommended to install the library within a Python virtual environment. This isolates project dependencies.

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

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

### Install eventsourcing with combined PostgreSQL and cryptography support

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

Combine options to install dependencies for both storing events in PostgreSQL and using cryptographic encryption. This installs packages for both 'postgres' and 'cryptography'.

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

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

### Install eventsourcing with PostgreSQL support

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

Install the library with the 'postgres' option to include dependencies for storing events with PostgreSQL, such as Psycopg v3 and its connection pool package.

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

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

### Run All Development Tasks

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

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

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

```

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

### Run Project Test Suite

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

Executes the project's test suite to ensure all components are functioning correctly after installation.

```bash
$ make test

```

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

### Install eventsourcing with version pinning

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

To avoid installing future incompatible releases, specify the major and minor version numbers using the '~=' operator. This allows for bug fixes while avoiding breaking changes.

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

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

### Install Poetry with pipx

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

Installs a specific version of Poetry using pipx. This is a prerequisite for managing project environments and dependencies.

```bash
$ make install-poetry

```

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

### Create and use PostgresProcessRecorder

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Shows how to create a PostgresProcessRecorder, create its table, insert events atomically with tracking, select recorded events, select notifications, and get the latest tracked position.

```python
from eventsourcing.postgres import PostgresProcessRecorder

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

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

# Insert stored events atomically with 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)
assert recorded_events[0] == stored_event

# Select notifications from the application sequence.
notifications = process_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

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

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

### Install eventsourcing with PyCryptodome support

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

Alternatively, install the library with the 'crypto' option to include PyCryptodome for storing cryptographically encrypted events. Ensure your application environment is configured for encryption.

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

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

### PostgreSQL FTS Environment Configuration and Setup

Source: https://eventsourcing.readthedocs.io/en/latest/topics/examples/fts-process.html

Configures environment variables for PostgreSQL persistence and includes setup/teardown logic for database tables.

```python
class TestWithPostgres(ContentManagementSystemTestCase):
    env: ClassVar[dict[str, str]] = {
        "PERSISTENCE_MODULE": "eventsourcing.postgres",
        "PROCESS_RECORDER_TOPIC": get_topic(PostgresFtsProcessRecorder),
        "POSTGRES_DBNAME": "eventsourcing",
        "POSTGRES_HOST": "127.0.0.1",
        "POSTGRES_PORT": "5432",
        "POSTGRES_USER": "eventsourcing",
        "POSTGRES_PASSWORD": "eventsourcing",
    }

    def setUp(self) -> None:
        drop_tables()
        super().setUp()

    def tearDown(self) -> None:
        super().tearDown()
        drop_tables()

    def test_system(self) -> None:
        super().test_system()
```

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

### Install eventsourcing with cryptography support

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

Install the library with the 'cryptography' option to include the Python cryptography package for storing cryptographically encrypted events. Ensure your application environment is configured for encryption.

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

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

### SQLiteAggregateRecorder Setup and Usage

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Demonstrates setting up and using SQLiteAggregateRecorder for persistent storage of aggregate events. This includes creating the SQLite datastore, constructing the recorder, creating the necessary table, and then inserting and selecting events.

```python
from eventsourcing.sqlite import SQLiteAggregateRecorder

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

# Construct aggregate recorder and create database table.
aggregate_recorder = SQLiteAggregateRecorder(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)
```

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

### Construct and Use Event Store

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Constructs an event store object with a mapper and recorder, then puts and gets 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]
```

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

### Use Aggregate Instance

Source: https://eventsourcing.readthedocs.io/en/latest/topics/domain.html

Demonstrates the usage of an aggregate instance after defining its projector methods. This example shows creating a Dog aggregate and adding a trick.

```python
dog = Dog.create()
dog.add_trick("roll over")
assert dog.tricks == ["roll over"], dog.tricks

```

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

### Store and retrieve domain events with EventStore

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Illustrates using the EventStore to store domain events using put() and retrieve them using get(). Requires a mapper and a recorder to be configured.

```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]
```

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

### Instantiate and Use DCBApplication

Source: https://eventsourcing.readthedocs.io/en/latest/topics/dcb.html

Illustrates how to construct a `DCBApplication` instance with specific persistence and mapping configurations, and then use its methods to perform domain operations like registering entities, updating data, enrolling students, and querying information.

```python
# Construct app to use MessagePack and in-memory persistence.
app = CourseSubscriptions(env={
    "PERSISTENCE_MODULE": "eventsourcing.dcb.popo",
    "MAPPER_TOPIC": "eventsourcing.dcb.msgpack:MessagePackMapper",
})

# Construct enduring objects.
student_id = app.register_student("Sara")
course_id = app.register_course("History")

# Update the student name using a vertical slice.
app.update_student_name(student_id, "Sara P")

# Enrol the student on the course using a group.
app.enrol_student_on_course(student_id, course_id)

# Query for student and course names.
assert "Sara P" in app.list_students_for_course(course_id)
assert "History" in app.list_courses_for_student(student_id)
```

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

### Initialize DCBRepository

Source: https://eventsourcing.readthedocs.io/en/latest/topics/dcb.html

Construct a DCBRepository with an event store, specifying the mapper and recorder implementations. This sets up the repository for interacting with the event store.

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

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

```

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

### Instantiate Wiki and Create a Page

Source: https://eventsourcing.readthedocs.io/en/latest/topics/application.html

Constructs a `Wiki` application object and creates a new page with specified name and body. This is the initial step to populate the wiki.

```python
wiki = Wiki()

wiki.create_page(name="Erth", body="Lorem ipsum...")
```

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

### Application.__init__

Source: https://eventsourcing.readthedocs.io/en/latest/topics/application.html

Initialises an application with an `InfrastructureFactory`, a `Mapper`, an `ApplicationRecorder`, an `EventStore`, a `Repository`, and a `LocalNotificationLog`.

```APIDOC
## Application.__init__

### Description
Initialises an application with an `InfrastructureFactory`, a `Mapper`, an `ApplicationRecorder`, an `EventStore`, a `Repository`, and a `LocalNotificationLog`.

### Method
POST (conceptual, as this is an SDK method)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- `_env` (Mapping[str, str] | None) - Optional - Environment variables to configure the application.
```

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

### Instantiate and Use DogSchool Application

Source: https://eventsourcing.readthedocs.io/en/latest/topics/application.html

Create an instance of the DogSchool application, register a new dog, add several tricks to it using command methods, and then retrieve the list of tricks using a query method. Asserts are used to verify the correctness of the retrieved 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"

```

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

### Create and use PostgresTrackingRecorder

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Demonstrates creating a PostgresTrackingRecorder, creating its table, inserting tracking information, and querying the latest tracked position and existence of tracking IDs.

```python
from eventsourcing.postgres import PostgresTrackingRecorder

# Construct tracking recorder and create table.
tracking_recorder = PostgresTrackingRecorder(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)
```

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

### select_notifications

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Retrieves a sequence of Notification objects from an application sequence based on start, limit, and stop IDs. Supports inclusive or exclusive start ID filtering.

```APIDOC
## select_notifications

### Description
Returns a list of Notification objects representing events from an application sequence. If inclusive_of_start is True (the default), the returned Notification objects will have IDs greater than or equal to start and less than or equal to stop. If inclusive_of_start is False, the Notification objects will have IDs greater than start and less than or equal to stop.

### Parameters
- **_start** (int | None) - The starting ID for the notification sequence.
- **_limit** (int) - The maximum number of notifications to retrieve.
- **_stop** (int | None) - The ending ID for the notification sequence.
- **_topics** (Sequence[str]) - A sequence of topics to filter notifications by.
- **_inclusive_of_start** (bool) - Whether to include the start ID in the range (default: True).
```

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

### Construct and Use PostgreSQL Event Store

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Demonstrates constructing an infrastructure factory, application recorder, mapper, and event store for PostgreSQL. It then shows how to put and retrieve domain events.

```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]

```

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

### ApplicationRecorder.select_notifications

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Returns a list of Notification objects representing events from an application sequence. Allows filtering by start, stop, and topics, with an option to include or exclude the start ID.

```APIDOC
## ApplicationRecorder.select_notifications

### Description
Returns a list of Notification objects representing events from an application sequence. If `inclusive_of_start` is True (the default), the returned Notification objects will have IDs greater than or equal to `start` and less than or equal to `stop`. If `inclusive_of_start` is False, the Notification objects will have IDs greater than `start` and less than or equal to `stop`.

### Method
select_notifications

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **start** (int | None) - Description for start parameter
- **limit** (int) - Description for limit parameter
- **stop** (int | None) - Optional. Description for stop parameter
- **topics** (Sequence[str]) - Optional. Defaults to (). Description for topics parameter
- **inclusive_of_start** (bool) - Optional. Defaults to True. Description for inclusive_of_start parameter

### Request Example
```python
recorder.select_notifications(start=10, limit=5, stop=20, topics=['user.created'], inclusive_of_start=False)
```

### Response
#### Success Response (Sequence[Notification])
- **Notification objects** - A sequence of Notification objects matching the query criteria.
```

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

### Create PostgreSQL Application Recorder and Table

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Initializes a PostgresApplicationRecorder by extending PostgresAggregateRecorder and creates its database table.

```python
from eventsourcing.postgres import PostgresApplicationRecorder

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

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

### select_notifications

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Retrieves a list of event notifications from a specified start point, with an optional limit and stop point. It can also filter notifications by topics and control inclusivity of the start point.

```APIDOC
## select_notifications

### Description
Returns a list of event notifications from ‘start’, limited by ‘limit’.

### Signature
`select_notifications(_start : int | None_, _limit : int_, _stop : int | None = None_, _topics : Sequence[str] = ()_, _*_ , _inclusive_of_start : bool = True_) → Sequence[Notification]`
```

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

### Get Environment Variable

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Use the get() method on an Environment object to retrieve a value for a key. It searches prefixed variables first, then standard names, returning None if not found.

```python
value = environ.get("PERSISTENCE_MODULE")

assert value == "eventsourcing.popo"

```

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

### Build Documentation

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

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

```bash
$ make docs

```

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

### Construct and Use ApplicationSubscription

Source: https://eventsourcing.readthedocs.io/en/latest/topics/projection.html

Demonstrates constructing an ApplicationSubscription, recording an event, and saving it. This snippet requires importing Application, Aggregate, and ApplicationSubscription.

```python
from uuid import UUID

from eventsourcing.application import Application
from eventsourcing.domain import Aggregate
from eventsourcing.projection import ApplicationSubscription

# Construct an application object.
app = Application[UUID]()

# Record an event.
aggregate = Aggregate()
app.save(aggregate)

```

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

### Assert notification IDs

Source: https://eventsourcing.readthedocs.io/en/latest/topics/application.html

Check that the `id` attribute of the selected notifications corresponds to the expected sequence, starting from the `start` value provided to `select()`. By default, the first notification has an ID of 1.

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

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

### Get Course

Source: https://eventsourcing.readthedocs.io/en/latest/topics/examples/dcb-enrolment-with-enduring-objects.html

Retrieves detailed information about a specific course.

```APIDOC
## get_course

### Description
Retrieves a course object.

### Method
Not applicable (Python method)

### Parameters
- **course_id** (CourseID) - Required - The ID of the course to retrieve.

### Returns
- **Course** - The course object with its details.
```

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

### Run All Development Tasks (Simplified)

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

A simplified command to run all essential development tasks.

```bash
$ make all

```

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

### Construct and Use Event Store with SQLite

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Construct infrastructure using a configured environment, then create an application recorder and event store. Assumes 'transcoder', 'domain_event', and 'id1' are defined.

```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]

```

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

### Get Student

Source: https://eventsourcing.readthedocs.io/en/latest/topics/examples/dcb-enrolment-with-enduring-objects.html

Retrieves detailed information about a specific student.

```APIDOC
## get_student

### Description
Retrieves a student object.

### Method
Not applicable (Python method)

### Parameters
- **student_id** (StudentID) - Required - The ID of the student to retrieve.

### Returns
- **Student** - The student object with their details.
```

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

### Order Command Method Execution

Source: https://eventsourcing.readthedocs.io/en/latest/topics/domain.html

Illustrates the execution of command methods on the Order class. Shows how calling pickup() before confirm() raises an exception, and how confirmation proceeds.

```python
# Create a new order.
order = Order("my order")
assert order.name == "my order"
assert order.confirmed_at is None
assert order.pickedup_at is None

# Can't pickup() before confirm().
try:
    order.pickup(datetime.now())
except AssertionError as e:
    assert e.args[0] == "Order is not confirmed"
    assert order.confirmed_at is None
    assert order.pickedup_at is None
else:
    raise Exception("shouldn't get here")

# Confirm the order.
order.confirm(datetime.now())
assert order.confirmed_at is not None
assert order.pickedup_at is None
```

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

### ConnectionUnavailableError

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Raised when a request to get a connection from a connection pool times out.

```APIDOC
_exception _eventsourcing.persistence.ConnectionUnavailableError[source]
    
    Bases: `OperationalError`, `TimeoutError`
    Raised when a request to get a connection from a connection pool times out.
```

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

### ShutDown Exception

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Exception raised when attempting to put or get items from a queue that has been shut down.

```APIDOC
## Exception ShutDown

Bases: `Exception`

Raised when put/get with shut-down queue.
```

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

### Update Locked Package Dependencies

Source: https://eventsourcing.readthedocs.io/en/latest/topics/installing.html

Updates the locked package dependencies to their latest compatible versions and installs them.

```bash
$ make update

```

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

### Create Initial Page and Index Aggregates

Source: https://eventsourcing.readthedocs.io/en/latest/topics/domain.html

Demonstrates creating a `Page` aggregate and an associated `Index` aggregate using the `Page`'s name and ID. This sets up the initial state for tracking a page by its name.

```python
page = Page.create(name="Erth")
index1 = Index.create(page.name, page.id)

```

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

### Create Timestamp for DomainEvent

Source: https://eventsourcing.readthedocs.io/en/latest/topics/domain.html

Use the static create_timestamp() method to get a timezone-aware datetime for the current time.

```python
timestamp = DomainEvent.create_timestamp()

assert isinstance(timestamp, datetime)
```

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

### Create a New Course Object

Source: https://eventsourcing.readthedocs.io/en/latest/topics/dcb.html

Demonstrates the creation of a 'Course' object with a name and maximum number of students. Asserts the correct initialization of these properties.

```python
course = Course(
    name="History",
    max_students=30,
)
assert course.name == "History"
assert course.max_students == 30
```

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

### Environment Class

Source: https://eventsourcing.readthedocs.io/en/latest/topics/domain.html

A dictionary-like class for handling environment variables, providing methods to get values with optional defaults.

```python
_class _eventsourcing.utils.Environment(_name : str = ''_, _env : Mapping[str, str] | None = None_)[source]
```

```python
get(___key : str_, _/_) → str | None[source]
```

```python
get(___key : str_, _/_, ___default : str_) → str[source]
```

```python
get(___key : str_, _/_, ___default : T_) → str | T[source]
```

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

### Create SQLite Process Recorder and Table

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Initializes an SQLiteProcessRecorder and creates its database table, extending SQLiteApplicationRecorder.

```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()
```

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

### Import Aggregate Base Class

Source: https://eventsourcing.readthedocs.io/en/latest/topics/domain.html

Import the base Aggregate class from the eventsourcing.domain module to start building event-sourced aggregates.

```python
from eventsourcing.domain import Aggregate
```

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

### SQLite Database URI Options

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Provides examples for configuring the SQLITE_DBNAME environment variable for different SQLite database connection types.

```python
# Configure SQLite database URI. Either use a file-based DB;
environ['SQLITE_DBNAME'] = '/path/to/your/sqlite-db'

# or use an in-memory DB with cache not shared, only works with single thread;
environ['SQLITE_DBNAME'] = ':memory:'

# or use an unnamed in-memory DB with shared cache, works with multiple threads;
environ['SQLITE_DBNAME'] = 'file::memory:?mode=memory&cache=shared'

```

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

### SQLiteApplicationRecorder Initialization

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Shows the initialization of SQLiteApplicationRecorder, which extends SQLiteAggregateRecorder. Note that this class does not implement the subscribe method.

```python
from eventsourcing.sqlite import SQLiteApplicationRecorder


```

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

### Implement DogSchool Application

Source: https://eventsourcing.readthedocs.io/en/latest/topics/examples/aggregate9.html

Implements the `DogSchool` application using `MsgspecApplication`. It handles dog registration, trick addition, and querying dog information, with snapshotting enabled.

```python
class DogSchool(MsgspecApplication):
    is_snapshotting_enabled = True
    snapshot_class = Snapshot

    def register_dog(self, name: str) -> UUID:
        event = register_dog(name)
        self.save(event)
        return event.originator_id

    def add_trick(self, dog_id: UUID, trick: str) -> None:
        dog = self.repository.get(dog_id, projector_func=project_dog)
        self.save(add_trick(dog, Trick(name=trick)))

    def get_dog(self, dog_id: UUID) -> dict[str, Any]:
        dog = self.repository.get(dog_id, projector_func=project_dog)
        return {
            "id": dog.id,
            "name": dog.name,
            "tricks": tuple([t.name for t in dog.tricks]),
            "created_on": dog.created_on,
            "modified_on": dog.modified_on,
        }

```

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

### Define Aggregate with Initialized Attributes

Source: https://eventsourcing.readthedocs.io/en/latest/topics/domain.html

Call the aggregate class with a name to initialize attributes. This example shows a simple Pig aggregate.

```python
pig = Pig("snowball")
assert pig.name == "snowball"
```

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

### Create SQLite Tracking Recorder and Table

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Initializes a SQLiteTrackingRecorder and creates its database table for tracking records.

```python
from eventsourcing.sqlite import SQLiteTrackingRecorder

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

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

### Cargo Events

Source: https://eventsourcing.readthedocs.io/en/latest/topics/examples/cargo-shipping.html

Events related to the Cargo aggregate, such as booking started, destination changed, route assigned, and handling events registered.

```APIDOC
## Cargo Events

### BookingStarted Event
- **Description**: Event indicating the start of a cargo booking.
- **Properties**:
  - **originator_id**: UUID
  - **originator_version**: int
  - **timestamp**: datetime
  - **origin**: Location
  - **destination**: Location
  - **arrival_deadline**: datetime

### DestinationChanged Event
- **Description**: Event indicating a change in the cargo's destination.
- **Properties**:
  - **originator_id**: UUID
  - **originator_version**: int
  - **timestamp**: datetime
  - **destination**: Location

### RouteAssigned Event
- **Description**: Event indicating that a route has been assigned to the cargo.
- **Properties**:
  - **originator_id**: UUID
  - **originator_version**: int
  - **timestamp**: datetime
  - **route**: Itinerary

### HandlingEventRegistered Event
- **Description**: Event indicating that a handling event has been registered for the cargo.
- **Properties**:
  - **originator_id**: UUID
  - **originator_version**: int
  - **timestamp**: datetime
  - **tracking_id**: UUID
  - **voyage_number**: str | None
  - **location**: Location
  - **handling_activity**: str
```

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

### Get Latest Aggregate State

Source: https://eventsourcing.readthedocs.io/en/latest/topics/application.html

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

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

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

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

### run

Source: https://eventsourcing.readthedocs.io/en/latest/topics/examples/content-management.html

Executes a command with provided arguments.

```APIDOC
## run

### Description
Executes a command with provided arguments.

### Signature
`run(_cmd: str, _a: str, _b: str) -> str`
```

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

### Create PostgreSQL Aggregate Recorder and Table

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Initializes a PostgresAggregateRecorder and creates its database table for storing events.

```python
from eventsourcing.postgres import PostgresAggregateRecorder

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

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

### Define Custom DateAsISO Transcoding

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Provides an example of creating a custom Transcoding subclass for Python date objects, encoding them as ISO strings.

```python
from datetime import date
from eventsourcing.persistence import Transcoding

class DateAsISO(Transcoding):
    type = date
    name = "date_iso"

    def encode(self, obj: date) -> str:
        return obj.isoformat()

    def decode(self, data: str) -> date:
        return date.fromisoformat(data)

transcoder.register(DateAsISO())
json_bytes = transcoder.encode(date1)
copy = transcoder.decode(json_bytes)
assert copy == date1
```

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

### Create SQLite Database and Application Recorder

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Initializes an in-memory SQLite database and an application recorder, then creates the necessary table for storing events.

```python
from eventsourcing.sqlite import SQLiteDatastore, 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()
```

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

### BaseProjectionRunner

Source: https://eventsourcing.readthedocs.io/en/latest/topics/projection.html

Abstract base class for projection runners. It handles the core logic for running a projection, including starting, stopping, and processing events.

```APIDOC
## class eventsourcing.projection.BaseProjectionRunner(projection: EventSourcedProjection[Any] | Projection[Any], application_class: type[TApplication], tracking_recorder: TrackingRecorder, topics: Sequence[str], env: Mapping[str, str] | None = None)

### Description
Initializes the BaseProjectionRunner.

### Parameters
- **projection** (EventSourcedProjection[Any] | Projection[Any]) - The projection to run.
- **application_class** (type[TApplication]) - The application class to use.
- **tracking_recorder** (TrackingRecorder) - The tracking recorder to use.
- **topics** (Sequence[str]) - The topics to subscribe to.
- **env** (Mapping[str, str] | None, optional) - Environment variables. Defaults to None.
```

```APIDOC
## stop() -> None

### Description
Sets the 'interrupted' event to stop the projection runner.

### Method
GET
```

```APIDOC
## run_forever(timeout: float | None = None) -> None

### Description
Blocks until timeout, or until the runner is stopped or errors. Re-raises any error otherwise exits normally.

### Parameters
- **timeout** (float | None, optional) - The timeout in seconds. Defaults to None.
```

```APIDOC
## wait(notification_id: int | None, timeout: float = 1.0) -> None

### Description
Blocks until timeout, or until the materialized view has recorded a tracking object that is greater than or equal to the given notification ID.

### Parameters
- **notification_id** (int | None) - The notification ID to wait for.
- **timeout** (float, optional) - The timeout in seconds. Defaults to 1.0.
```

```APIDOC
## __enter__() -> Self

### Description
Enter the runtime context related to this object.
```

```APIDOC
## __exit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None

### Description
Calls stop() and waits for the event-processing thread to exit.
```

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

### Define DCBApplication with Command Methods

Source: https://eventsourcing.readthedocs.io/en/latest/topics/dcb.html

Shows how to define a custom application class inheriting from `DCBApplication`. This class includes methods for registering students and courses, updating student names, enrolling students, and listing related entities.

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


class CourseSubscriptions(DCBApplication):
    def register_student(self, name: str) -> str:
        student = Student(name=name, max_courses=5)
        self.repository.save(student)
        return student.id

    def update_student_name(self, student_id: str, new_name: str) -> None:
        self.do(UpdateStudentName(student_id, new_name))

    def register_course(self, name: str) -> str:
        course = Course(name=name, max_students=30)
        self.repository.save(course)
        return course.id

    def enrol_student_on_course(self, student_id: str, course_id: str) -> None:
        group = self.repository.get_group(StudentAndCourse, student_id, course_id)
        group.student_joins_course()
        self.repository.save(group)

    def list_courses_for_student(self, student_id: str) -> list[str]:
        student: Student = self.repository.get(student_id)
        return [c.name for c in self.repository.get_many(*student.course_ids)]

    def list_students_for_course(self, course_id: str) -> list[str]:
        course: Course = self.repository.get(course_id)
        return [s.name for s in self.repository.get_many(*course.student_ids)]
```

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

### BookingStarted Event

Source: https://eventsourcing.readthedocs.io/en/latest/topics/examples/cargo-shipping.html

Event class representing the start of a cargo booking, containing essential details like origin, destination, and arrival deadline.

```python
    class BookingStarted(Aggregate.Created):
        origin: Location
        destination: Location
        arrival_deadline: datetime
```

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

### Run Projection with Default Persistence

Source: https://eventsourcing.readthedocs.io/en/latest/topics/projection.html

Demonstrates how to run a projection using the ProjectionRunner with default persistence. It includes a signal handler to stop the runner after a delay.

```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=Application,
    view_class=MyPOPOMaterialisedView,
    projection_class=AggregateEventProjection,
    env={},
) as projection_runner:

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

    # Run until interrupted.
    projection_runner.run_forever()

```

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

### New Cargo Booking

Source: https://eventsourcing.readthedocs.io/en/latest/topics/examples/cargo-shipping.html

Class method to create a new Cargo aggregate instance and start the booking process by triggering a BookingStarted event.

```python
    @classmethod
    def new_booking(
        cls,
        origin: Location,
        destination: Location,
        arrival_deadline: datetime,
    ) -> Cargo:
        return cls._create(
            event_class=cls.BookingStarted,
            id=uuid4(),
            origin=origin,
            destination=destination,
            arrival_deadline=arrival_deadline,
        )
```

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

### TestSubmitCart - Sufficient Inventory

Source: https://eventsourcing.readthedocs.io/en/latest/topics/examples/shop-vertical.html

Tests submitting a cart with sufficient inventory after adding a product to the shop. This test setup is incomplete in the provided source.

```python
    def test_submit_cart_sufficient_inventory_after_item_added(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()

        AddProductToShop(
            product_id=product_id,
            name="",
            description="",
            price=Decimal(100),
        ).execute()

```

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

### Verify Product Details in Shop

Source: https://eventsourcing.readthedocs.io/en/latest/topics/examples/shop-vertical.html

Asserts the properties of listed products, such as name, description, price, and inventory. Ensure Decimal is imported for price assertions.

```python
self.assertEqual(products[0].name, "Coffee")
        self.assertEqual(products[0].description, "A very nice coffee")
        self.assertEqual(products[0].price, Decimal("5.99"))
        self.assertEqual(products[0].inventory, 2)
        self.assertEqual(products[1].id, product_id2)
        self.assertEqual(products[1].name, "Tea")
        self.assertEqual(products[1].description, "A very nice tea")
        self.assertEqual(products[1].price, Decimal("3.99"))
        self.assertEqual(products[1].inventory, 0)
```

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

### Advance and Execute Decisions with Repository

Source: https://eventsourcing.readthedocs.io/en/latest/topics/dcb.html

Demonstrates how to use the repository's `advance` method to apply decisions to a perspective and then execute them. This is useful for updating a slice to its current state before executing.

```python
student = repository.get(student.id)
course = repository.get(course.id)
assert course.id in student.course_ids
assert student.id in course.student_ids

update_student_name = UpdateStudentName(student_id=student.id, new_name="Sara P")
assert update_student_name.name == ""
assert update_student_name.student_was_registered is False

repository.advance(update_student_name)
assert update_student_name.name == "Sara"
assert update_student_name.student_was_registered is True

update_student_name.execute()
assert update_student_name.name == "Sara P"
assert update_student_name.student_was_registered is True

repository.save(update_student_name)
```

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

### Inspect first notification details

Source: https://eventsourcing.readthedocs.io/en/latest/topics/application.html

Examine the `topic` and `originator_id` of the first selected notification to understand the event it represents. This example checks for a 'Dog.Created' topic.

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

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

### max_notification_id

Source: https://eventsourcing.readthedocs.io/en/latest/topics/persistence.html

Retrieves the highest notification ID currently stored. This is useful for determining the latest event or for setting a starting point for fetching new notifications.

```APIDOC
## max_notification_id

### Description
Returns the maximum notification ID.

### Signature
`max_notification_id() → int | None`
```