Projections¶

Projections provide a structured way to read and write data efficiently. They handle event collection, logging, event bus publishing, and transactional commit/rollback automatically.

Data Models¶

from pydantic import BaseModel

class UserSearch(BaseModel):
    query: str
    page: int = 1

class UserUpdate(BaseModel):
    user_id: str
    name: str
    email: str

ReadProjection¶

from aod.infrastructure import ReadProjection

ReadProjection(ReadProjectionBase) — Base class for synchronous read projections.

Constructor¶

ReadProjection(**fields)

Methods¶

read Parameters¶

Auto-Wrapping Behavior¶

When read() is called:

1. Events are collected via EventCollector during execution.
2. On success: events are logged on each declared logger, events are published on each declared event bus.
3. On failure: the exception is logged on each declared logger and re-raised.

Example¶

from pydantic import BaseModel
from aod.infrastructure import ReadProjection

class UserSearch(BaseModel):
    user_id: str

class UserListProjection(ReadProjection):
    session: PostgresSession

    def read(self, model: UserSearch) -> list[User]:
        rows = self.session.query(
            "SELECT * FROM users WHERE id = :id",
            {"id": model.user_id},
        )
        return [User(**row) for row in rows]

WriteProjection¶

from aod.infrastructure import WriteProjection

WriteProjection(WriteProjectionBase) — Base class for synchronous write projections.

Constructor¶

WriteProjection(**fields)

Methods¶

write Parameters¶

Auto-Wrapping Behavior¶

When write() is called:

1. A CommitContext is set (enabling session.commit()).
2. Events are collected via EventCollector during execution.
3. On success: events are logged on each declared logger, events are published on each declared event bus.
4. On failure: session.rollback() is called (if session is dirty), the exception is logged on each declared logger and re-raised.
5. The CommitContext is always reset in the finally block.

Example¶

from pydantic import BaseModel
from aod.infrastructure import WriteProjection

class UpdateUserInput(BaseModel):
    user_id: str
    name: str

class UserUpdateProjection(WriteProjection):
    session: PostgresSession

    def write(self, model: UpdateUserInput) -> None:
        self.session.execute(
            "UPDATE users SET name = :name WHERE id = :id",
            {"name": model.name, "id": model.user_id},
        )

Projection¶

from aod.infrastructure import Projection

Projection(ReadProjection, WriteProjection) — Combines both read and write capabilities.

Constructor¶

Projection(**fields)

Methods¶

Includes both read(self, model: Any) -> Any and write(self, model: Any) -> Any.

Async Variants¶

Constructor¶

Async variants accept the same fields as their sync counterparts. Session fields must use concrete types (e.g., session: AsyncPostgresSession). Multiple session fields are supported.

All async variants expose the same methods but as async: - async read(self, *args, **kwargs) -> Any - async write(self, *args, **kwargs) -> Any

Field Validation¶

Projections enforce these rules at class creation time:

1. Concrete session types — Session fields must use concrete types (e.g., session: PostgresSession), never Session | None.
2. No HandlerProtocol — Fields typed as HandlerProtocol or its subclasses raise InvalidUseCasePortFieldError.
3. Multiple sessions allowed — Projections can declare multiple session fields with different types.
4. No HandlerProtocol — Fields typed as HandlerProtocol or its subclasses raise InvalidUseCasePortFieldError.

Event Collection¶

Events emitted during read() or write() are automatically collected:

• Collected events are stored on self.events after execution completes.
• Events are published on the event bus after a successful operation.
• Write projections require a successful commit before events are published.

Testing with Spy Sessions¶

from aod.testing.doubles import spy_session
from pydantic import BaseModel

class UserSearch(BaseModel):
    user_id: int

StubMySession = spy_session(MySession)

class MyReadProjection(ReadProjection):
    session: StubMySession

    def read(self, model: UserSearch) -> Any:
        result = self.session.query("SELECT * FROM users")
        ...

proj = MyReadProjection(session=StubMySession())
result = proj.read(UserSearch(user_id=42))
assert proj.session.query.called

Next Steps¶