orchestrata_ultimate.core.base.repository

Reference › orchestrata_ultimate › core › base › repository

The persistence contract for entities. BaseRepository[T] is a Protocol generic over BaseEntity subclasses, with four async operations: get(entity_id) returns the entity or None, save(entity) creates or updates (the docstring says it calls touch() first), delete(entity_id) returns a boolean, and list(offset, limit) pages with a default limit of 100. The protocol leaves the storage backend to the implementation (Redis, graph, SQLite or memory). It declares a _lock attribute of type threading.Lock, which implementations use to guard connection-pool access under free-threaded Python 3.14t.

At a glance


Source	`src/orchestrata_ultimate/core/base/repository.py`
Lines of code	80
Classes	1
Functions	0
Imports (internal)	`orchestrata_ultimate.core.base.entity`
Imported by	`orchestrata_ultimate.core.base`
Tests that import it	`tests/core/test_base_repository.py`, `tests/generated_core/test_base_repository_generated.py`

Class diagram

Dependency diagram

API

Classes

`BaseRepository` (protocol)

Bases: Protocol · Line 14

Protocol for entity repositories.

Attribute	Type	Default
`_lock`	`threading.Lock`

Method	Signature	Summary
`__init__`	`__init__() -> None`	Initialize with a threading lock.
`get`	`async get(entity_id: str) -> T \| None`	Retrieve entity by ID.
`save`	`async save(entity: T) -> T`	Create or update an entity.
`delete`	`async delete(entity_id: str) -> bool`	Delete entity by ID.
`list`	`async list(offset: int = 0, limit: int = 100) -> list[T]`	List entities with pagination.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/base/repository.py

# mypy: ignore-errors
"""BaseRepository — async CRUD interface for entity persistence."""

from __future__ import annotations

import threading
from typing import Protocol, TypeVar

from orchestrata_ultimate.core.base.entity import BaseEntity

T = TypeVar("T", bound=BaseEntity)


class BaseRepository[T: BaseEntity](Protocol):
    """Protocol for entity repositories.

    Provides async CRUD interface with offset/limit pagination.
    Concrete implementations handle the storage backend
    (Redis, Neo4j, SQLite, in-memory).

    Thread safety: implementations must use self._lock for
    connection pool access under Python 3.14t free-threaded.
    """

    _lock: threading.Lock

    def __init__(self) -> None:
        """Initialize with a threading lock."""
        self._lock = threading.Lock()

    async def get(self, entity_id: str) -> T | None:
        """Retrieve entity by ID.

        Args:
            entity_id: Entity UUID string.

        Returns:
            Entity instance or None if not found.
        """
        ...

    async def save(self, entity: T) -> T:
        """Create or update an entity.

        Calls entity.touch() before persisting.

        Args:
            entity: Entity to save.

        Returns:
            Saved entity (may have updated fields).
        """
        ...

    async def delete(self, entity_id: str) -> bool:
        """Delete entity by ID.

        Args:
            entity_id: Entity UUID string.

        Returns:
            True if deleted, False if not found.
        """
        ...

    async def list(
        self,
        offset: int = 0,
        limit: int = 100,
    ) -> list[T]:
        """List entities with pagination.

        Args:
            offset: Number of items to skip.
            limit: Maximum items to return.

        Returns:
            List of entities.
        """
        ...