Skip to main content

RepositoryProtocol

Protocol defining the contract for data repositories.

This is the primary port for data persistence in the hexagonal architecture.
All repository adapters must implement this interface.

Type parameter T represents the entity type being managed.

Features:
- Generic CRUD operations (get, save, delete)
- Existence and count queries
- Paginated listing with filtering and sorting
- Bulk operations for performance
- Optional Optimistic Concurrency Control (OCC) via version parameter

Example implementation:
    class SQLAlchemyRepository(RepositoryProtocol[User]):
        async def get(self, id: UUID) -> User | None:
            ...

Source: repository.py

Methods​

get​

async def get(self, id: UUID) -> T | None

Retrieve an entity by its unique identifier.

    Args:
        id: The unique identifier of the entity

    Returns:
        The entity if found, None otherwise

save​

async def save(self, entity: T, version: int | None = None) -> T

Persist an entity (insert or update).

    If the entity has an id, it will be updated. Otherwise, inserted.
    For Optimistic Concurrency Control, pass the expected version.

    Args:
        entity: The entity to save
        version: Expected version for OCC (raises VersionConflictError if mismatch)

    Returns:
        The saved entity with updated fields (id, modified, version, etc.)

    Raises:
        VersionConflictError: If version doesn't match (OCC)

delete​

async def delete(self, id: UUID) -> None

Delete an entity by its unique identifier.

    Args:
        id: The unique identifier of the entity to delete

    Raises:
        EntityNotFoundError: If entity doesn't exist

exists​

async def exists(self, id: UUID) -> bool

Check if an entity exists by its identifier.

    More efficient than get() when you only need existence check.

    Args:
        id: The unique identifier to check

    Returns:
        True if entity exists, False otherwise

count​

async def count(self, filters: list[FilterSpec] | None = None) -> int

Count entities matching the given filters.

    Args:
        filters: Optional list of filter conditions

    Returns:
        Count of matching entities

list​

async def list(self,
    filters: list[FilterSpec] | None = None,
    order_by: list[OrderSpec] | None = None,
    limit: int = 20,
    offset: int = 0,
) -> PaginatedResult[T]

List entities with filtering, sorting, and pagination.

    Args:
        filters: Optional list of filter conditions
        order_by: Optional list of sort specifications
        limit: Maximum number of items to return (default: 20)
        offset: Number of items to skip (default: 0)

    Returns:
        PaginatedResult containing items and pagination metadata

bulk_save​

async def bulk_save(self, entities: Sequence[T]) -> Sequence[T]

Save multiple entities in a single operation.

    More efficient than calling save() multiple times.
    All entities are saved within a single transaction.

    Args:
        entities: List of entities to save

    Returns:
        List of saved entities with updated fields