docs/standards/mcp/templates/app/infra/db/repos/memory.py

"""# Summary

In-memory integer-id repository — the zero-dependency persistence adapter.

---

One process-local dictionary per resource, keyed by auto-incrementing integer
``id``. The wire contract (``openapi.yaml``) declares integer primary keys,
so this adapter speaks integers end-to-end and is the default backend
(``REPO_BACKEND=memory``).

The method surface mirrors the mongo-ops repository duck-type consumed by
``infra.adapters.service.CrudsServiceAdaptor``: ``create``, ``get_by_id``,
``get_many``, ``update``, ``patch``, ``delete``.

Notes:
    Not thread-safe and not durable — deliberately. Swap to the Mongo
    adapter via ``REPO_BACKEND=mongo`` without touching domain code.
"""
from typing import Any

from pydantic import BaseModel


class IntIdMemoryRepository:
    """Dictionary-backed repository with auto-increment integer ids.

    Attributes:
        document_cls (type[BaseModel]): Document model instantiated for
            every stored record.
    """

    def __init__(self, document_cls: type[BaseModel]):
        """Initialize an empty store for ``document_cls``.

        Args:
            document_cls (type[BaseModel]): Document model bound to this repo.
        """
        self.document_cls = document_cls
        self._docs: dict[int, BaseModel] = {}
        self._seq = 0

    def _to_doc(self, data: Any) -> BaseModel:
        """Coerce ``data`` into a document instance.

        Args:
            data (Any): Document instance or field mapping.

        Returns:
            BaseModel: Document built from ``data``.
        """
        if isinstance(data, self.document_cls):
            return data
        return self.document_cls(**dict(data))

    async def create(self, data: Any) -> BaseModel:
        """Persist a new document with a freshly allocated integer id.

        Args:
            data (Any): Document instance or field mapping (id ignored).

        Returns:
            BaseModel: The stored document with ``id`` assigned.
        """
        doc = self._to_doc(data)
        self._seq += 1
        doc.id = self._seq
        self._docs[doc.id] = doc
        return doc

    async def get_by_id(self, id: int) -> BaseModel | None:
        """Fetch one document by integer id.

        Args:
            id (int): Document identifier.

        Returns:
            BaseModel | None: The stored document, or ``None`` when absent.
        """
        return self._docs.get(int(id))

    async def get_many(self, filter: dict | None = None, **_: Any) -> list[BaseModel]:
        """List documents, optionally filtered by top-level equality.

        Args:
            filter (dict | None): Equality filter on document fields.

        Returns:
            list[BaseModel]: Matching documents in insertion order.
        """
        docs = list(self._docs.values())
        if filter:
            docs = [d for d in docs if all(getattr(d, k, None) == v for k, v in filter.items())]
        return docs

    async def update(self, id: int, data: Any) -> BaseModel | None:
        """Replace a document by integer id.

        Args:
            id (int): Document identifier.
            data (Any): Replacement document or field mapping.

        Returns:
            BaseModel | None: The stored document, or ``None`` when absent.
        """
        if int(id) not in self._docs:
            return None
        doc = self._to_doc(data)
        doc.id = int(id)
        self._docs[doc.id] = doc
        return doc

    async def patch(self, id: int, data: dict) -> BaseModel | None:
        """Partially update a document by integer id.

        Args:
            id (int): Document identifier.
            data (dict): Fields to merge into the stored document.

        Returns:
            BaseModel | None: The stored document, or ``None`` when absent.
        """
        current = self._docs.get(int(id))
        if current is None:
            return None
        merged = current.model_dump()
        merged.update(data)
        doc = self.document_cls(**merged)
        doc.id = int(id)
        self._docs[doc.id] = doc
        return doc

    async def delete(self, id: int) -> bool:
        """Delete a document by integer id.

        Args:
            id (int): Document identifier.

        Returns:
            bool: ``True`` when a document was removed.
        """
        return self._docs.pop(int(id), None) is not None

    async def delete_many(self, filter: dict | None = None) -> int:
        """Delete every document (optionally filtered by equality).

        Args:
            filter (dict | None): Equality filter on document fields.

        Returns:
            int: Number of documents removed.
        """
        victims = await self.get_many(filter)
        for doc in victims:
            self._docs.pop(doc.id, None)
        return len(victims)

    def clear(self) -> None:
        """Remove every document and reset the id counter."""
        self._docs.clear()
        self._seq = 0