diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/repository')
22 files changed, 1626 insertions, 0 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/repository/__init__.py b/venv/lib/python3.11/site-packages/litestar/repository/__init__.py new file mode 100644 index 0000000..62e3f83 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/__init__.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +from .abc import AbstractAsyncRepository, AbstractSyncRepository +from .exceptions import ConflictError, NotFoundError, RepositoryError +from .filters import FilterTypes + +__all__ = ( + "AbstractAsyncRepository", + "AbstractSyncRepository", + "ConflictError", + "FilterTypes", + "NotFoundError", + "RepositoryError", +) diff --git a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/__init__.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..42bf9be --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/__init__.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_exceptions.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_exceptions.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..0f48627 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_exceptions.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_filters.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_filters.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..2f16cb4 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_filters.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/exceptions.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/exceptions.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..3c9f01e --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/exceptions.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/filters.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/filters.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..87d90db --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/filters.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/handlers.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/handlers.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..abe227b --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/handlers.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/_exceptions.py b/venv/lib/python3.11/site-packages/litestar/repository/_exceptions.py new file mode 100644 index 0000000..1c2b7be --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/_exceptions.py @@ -0,0 +1,15 @@ +from __future__ import annotations # pragma: no cover + +__all__ = ("ConflictError", "NotFoundError", "RepositoryError") # pragma: no cover + + +class RepositoryError(Exception): # pragma: no cover + """Base repository exception type.""" + + +class ConflictError(RepositoryError): # pragma: no cover + """Data integrity error.""" + + +class NotFoundError(RepositoryError): # pragma: no cover + """An identity does not exist.""" diff --git a/venv/lib/python3.11/site-packages/litestar/repository/_filters.py b/venv/lib/python3.11/site-packages/litestar/repository/_filters.py new file mode 100644 index 0000000..f6b787e --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/_filters.py @@ -0,0 +1,117 @@ +"""Collection filter datastructures.""" + +from __future__ import annotations + +from collections import abc # noqa: TCH003 +from dataclasses import dataclass +from datetime import datetime # noqa: TCH003 +from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar + +if TYPE_CHECKING: + from typing_extensions import TypeAlias + +T = TypeVar("T") + +__all__ = ( + "BeforeAfter", + "CollectionFilter", + "FilterTypes", + "LimitOffset", + "OrderBy", + "SearchFilter", + "NotInCollectionFilter", + "OnBeforeAfter", + "NotInSearchFilter", +) + + +FilterTypes: TypeAlias = "BeforeAfter | OnBeforeAfter | CollectionFilter[Any] | LimitOffset | OrderBy | SearchFilter | NotInCollectionFilter[Any] | NotInSearchFilter" +"""Aggregate type alias of the types supported for collection filtering.""" + + +@dataclass +class BeforeAfter: + """Data required to filter a query on a ``datetime`` column.""" + + field_name: str + """Name of the model attribute to filter on.""" + before: datetime | None + """Filter results where field earlier than this.""" + after: datetime | None + """Filter results where field later than this.""" + + +@dataclass +class OnBeforeAfter: + """Data required to filter a query on a ``datetime`` column.""" + + field_name: str + """Name of the model attribute to filter on.""" + on_or_before: datetime | None + """Filter results where field is on or earlier than this.""" + on_or_after: datetime | None + """Filter results where field on or later than this.""" + + +@dataclass +class CollectionFilter(Generic[T]): + """Data required to construct a ``WHERE ... IN (...)`` clause.""" + + field_name: str + """Name of the model attribute to filter on.""" + values: abc.Collection[T] + """Values for ``IN`` clause.""" + + +@dataclass +class NotInCollectionFilter(Generic[T]): + """Data required to construct a ``WHERE ... NOT IN (...)`` clause.""" + + field_name: str + """Name of the model attribute to filter on.""" + values: abc.Collection[T] + """Values for ``NOT IN`` clause.""" + + +@dataclass +class LimitOffset: + """Data required to add limit/offset filtering to a query.""" + + limit: int + """Value for ``LIMIT`` clause of query.""" + offset: int + """Value for ``OFFSET`` clause of query.""" + + +@dataclass +class OrderBy: + """Data required to construct a ``ORDER BY ...`` clause.""" + + field_name: str + """Name of the model attribute to sort on.""" + sort_order: Literal["asc", "desc"] = "asc" + """Sort ascending or descending""" + + +@dataclass +class SearchFilter: + """Data required to construct a ``WHERE field_name LIKE '%' || :value || '%'`` clause.""" + + field_name: str + """Name of the model attribute to sort on.""" + value: str + """Values for ``LIKE`` clause.""" + ignore_case: bool | None = False + """Should the search be case insensitive.""" + + +@dataclass +class NotInSearchFilter: + """Data required to construct a ``WHERE field_name NOT LIKE '%' || :value || '%'`` clause.""" + + field_name: str + """Name of the model attribute to search on.""" + value: str + """Values for ``NOT LIKE`` clause.""" + ignore_case: bool | None = False + """Should the search be case insensitive.""" diff --git a/venv/lib/python3.11/site-packages/litestar/repository/abc/__init__.py b/venv/lib/python3.11/site-packages/litestar/repository/abc/__init__.py new file mode 100644 index 0000000..def6bb4 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/abc/__init__.py @@ -0,0 +1,7 @@ +from ._async import AbstractAsyncRepository +from ._sync import AbstractSyncRepository + +__all__ = ( + "AbstractAsyncRepository", + "AbstractSyncRepository", +) diff --git a/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/__init__.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..710a74b --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/__init__.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_async.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_async.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..24217cc --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_async.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_sync.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_sync.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..7a69407 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_sync.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/abc/_async.py b/venv/lib/python3.11/site-packages/litestar/repository/abc/_async.py new file mode 100644 index 0000000..85ca139 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/abc/_async.py @@ -0,0 +1,303 @@ +from __future__ import annotations + +from abc import ABCMeta, abstractmethod +from typing import TYPE_CHECKING, Any, Generic, TypeVar + +from litestar.repository.exceptions import NotFoundError + +if TYPE_CHECKING: + from litestar.repository.filters import FilterTypes + +T = TypeVar("T") +CollectionT = TypeVar("CollectionT") + + +class AbstractAsyncRepository(Generic[T], metaclass=ABCMeta): + """Interface for persistent data interaction.""" + + model_type: type[T] + """Type of object represented by the repository.""" + id_attribute: Any = "id" + """Name of the primary identifying attribute on :attr:`model_type`.""" + + def __init__(self, **kwargs: Any) -> None: + """Repository constructors accept arbitrary kwargs.""" + super().__init__(**kwargs) + + @abstractmethod + async def add(self, data: T) -> T: + """Add ``data`` to the collection. + + Args: + data: Instance to be added to the collection. + + Returns: + The added instance. + """ + + @abstractmethod + async def add_many(self, data: list[T]) -> list[T]: + """Add multiple ``data`` to the collection. + + Args: + data: Instances to be added to the collection. + + Returns: + The added instances. + """ + + @abstractmethod + async def count(self, *filters: FilterTypes, **kwargs: Any) -> int: + """Get the count of records returned by a query. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Instance attribute value filters. + + Returns: + The count of instances + """ + + @abstractmethod + async def delete(self, item_id: Any) -> T: + """Delete instance identified by ``item_id``. + + Args: + item_id: Identifier of instance to be deleted. + + Returns: + The deleted instance. + + Raises: + NotFoundError: If no instance found identified by ``item_id``. + """ + + @abstractmethod + async def delete_many(self, item_ids: list[Any]) -> list[T]: + """Delete multiple instances identified by list of IDs ``item_ids``. + + Args: + item_ids: list of Identifiers to be deleted. + + Returns: + The deleted instances. + """ + + @abstractmethod + async def exists(self, *filters: FilterTypes, **kwargs: Any) -> bool: + """Return true if the object specified by ``kwargs`` exists. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Identifier of the instance to be retrieved. + + Returns: + True if the instance was found. False if not found. + + """ + + @abstractmethod + async def get(self, item_id: Any, **kwargs: Any) -> T: + """Get instance identified by ``item_id``. + + Args: + item_id: Identifier of the instance to be retrieved. + **kwargs: Additional arguments + + Returns: + The retrieved instance. + + Raises: + NotFoundError: If no instance found identified by ``item_id``. + """ + + @abstractmethod + async def get_one(self, **kwargs: Any) -> T: + """Get an instance specified by the ``kwargs`` filters if it exists. + + Args: + **kwargs: Instance attribute value filters. + + Returns: + The retrieved instance. + + Raises: + NotFoundError: If no instance found identified by ``kwargs``. + """ + + @abstractmethod + async def get_or_create(self, **kwargs: Any) -> tuple[T, bool]: + """Get an instance specified by the ``kwargs`` filters if it exists or create it. + + Args: + **kwargs: Instance attribute value filters. + + Returns: + A tuple that includes the retrieved or created instance, and a boolean on whether the record was created or not + """ + + @abstractmethod + async def get_one_or_none(self, **kwargs: Any) -> T | None: + """Get an instance if it exists or None. + + Args: + **kwargs: Instance attribute value filters. + + Returns: + The retrieved instance or None. + """ + + @abstractmethod + async def update(self, data: T) -> T: + """Update instance with the attribute values present on ``data``. + + Args: + data: An instance that should have a value for :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` that exists in the + collection. + + Returns: + The updated instance. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + + @abstractmethod + async def update_many(self, data: list[T]) -> list[T]: + """Update multiple instances with the attribute values present on instances in ``data``. + + Args: + data: A list of instance that should have a value for :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` that exists in the + collection. + + Returns: + a list of the updated instances. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + + @abstractmethod + async def upsert(self, data: T) -> T: + """Update or create instance. + + Updates instance with the attribute values present on ``data``, or creates a new instance if + one doesn't exist. + + Args: + data: Instance to update existing, or be created. Identifier used to determine if an + existing instance exists is the value of an attribute on ``data`` named as value of + :attr:`id_attribute <AbstractAsyncRepository.id_attribute>`. + + Returns: + The updated or created instance. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + + @abstractmethod + async def upsert_many(self, data: list[T]) -> list[T]: + """Update or create multiple instances. + + Update instances with the attribute values present on ``data``, or create a new instance if + one doesn't exist. + + Args: + data: Instances to update or created. Identifier used to determine if an + existing instance exists is the value of an attribute on ``data`` named as value of + :attr:`id_attribute <AbstractAsyncRepository.id_attribute>`. + + Returns: + The updated or created instances. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + + @abstractmethod + async def list_and_count(self, *filters: FilterTypes, **kwargs: Any) -> tuple[list[T], int]: + """List records with total count. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Instance attribute value filters. + + Returns: + a tuple containing The list of instances, after filtering applied, and a count of records returned by query, ignoring pagination. + """ + + @abstractmethod + async def list(self, *filters: FilterTypes, **kwargs: Any) -> list[T]: + """Get a list of instances, optionally filtered. + + Args: + *filters: filters for specific filtering operations + **kwargs: Instance attribute value filters. + + Returns: + The list of instances, after filtering applied + """ + + @abstractmethod + def filter_collection_by_kwargs(self, collection: CollectionT, /, **kwargs: Any) -> CollectionT: + """Filter the collection by kwargs. + + Has ``AND`` semantics where multiple kwargs name/value pairs are provided. + + Args: + collection: the objects to be filtered + **kwargs: key/value pairs such that objects remaining in the collection after filtering + have the property that their attribute named ``key`` has value equal to ``value``. + + + Returns: + The filtered objects + + Raises: + RepositoryError: if a named attribute doesn't exist on :attr:`model_type <AbstractAsyncRepository.model_type>`. + """ + + @staticmethod + def check_not_found(item_or_none: T | None) -> T: + """Raise :class:`NotFoundError` if ``item_or_none`` is ``None``. + + Args: + item_or_none: Item (:class:`T <T>`) to be tested for existence. + + Returns: + The item, if it exists. + """ + if item_or_none is None: + raise NotFoundError("No item found when one was expected") + return item_or_none + + @classmethod + def get_id_attribute_value(cls, item: T | type[T], id_attribute: str | None = None) -> Any: + """Get value of attribute named as :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` on ``item``. + + Args: + item: Anything that should have an attribute named as :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` value. + id_attribute: Allows customization of the unique identifier to use for model fetching. + Defaults to `None`, but can reference any surrogate or candidate key for the table. + + Returns: + The value of attribute on ``item`` named as :attr:`id_attribute <AbstractAsyncRepository.id_attribute>`. + """ + return getattr(item, id_attribute if id_attribute is not None else cls.id_attribute) + + @classmethod + def set_id_attribute_value(cls, item_id: Any, item: T, id_attribute: str | None = None) -> T: + """Return the ``item`` after the ID is set to the appropriate attribute. + + Args: + item_id: Value of ID to be set on instance + item: Anything that should have an attribute named as :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` value. + id_attribute: Allows customization of the unique identifier to use for model fetching. + Defaults to `None`, but can reference any surrogate or candidate key for the table. + + Returns: + Item with ``item_id`` set to :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` + """ + setattr(item, id_attribute if id_attribute is not None else cls.id_attribute, item_id) + return item diff --git a/venv/lib/python3.11/site-packages/litestar/repository/abc/_sync.py b/venv/lib/python3.11/site-packages/litestar/repository/abc/_sync.py new file mode 100644 index 0000000..d667fc2 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/abc/_sync.py @@ -0,0 +1,305 @@ +# Do not edit this file directly. It has been autogenerated from +# litestar/repository/abc/_async.py +from __future__ import annotations + +from abc import ABCMeta, abstractmethod +from typing import TYPE_CHECKING, Any, Generic, TypeVar + +from litestar.repository.exceptions import NotFoundError + +if TYPE_CHECKING: + from litestar.repository.filters import FilterTypes + +T = TypeVar("T") +CollectionT = TypeVar("CollectionT") + + +class AbstractSyncRepository(Generic[T], metaclass=ABCMeta): + """Interface for persistent data interaction.""" + + model_type: type[T] + """Type of object represented by the repository.""" + id_attribute: Any = "id" + """Name of the primary identifying attribute on :attr:`model_type`.""" + + def __init__(self, **kwargs: Any) -> None: + """Repository constructors accept arbitrary kwargs.""" + super().__init__(**kwargs) + + @abstractmethod + def add(self, data: T) -> T: + """Add ``data`` to the collection. + + Args: + data: Instance to be added to the collection. + + Returns: + The added instance. + """ + + @abstractmethod + def add_many(self, data: list[T]) -> list[T]: + """Add multiple ``data`` to the collection. + + Args: + data: Instances to be added to the collection. + + Returns: + The added instances. + """ + + @abstractmethod + def count(self, *filters: FilterTypes, **kwargs: Any) -> int: + """Get the count of records returned by a query. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Instance attribute value filters. + + Returns: + The count of instances + """ + + @abstractmethod + def delete(self, item_id: Any) -> T: + """Delete instance identified by ``item_id``. + + Args: + item_id: Identifier of instance to be deleted. + + Returns: + The deleted instance. + + Raises: + NotFoundError: If no instance found identified by ``item_id``. + """ + + @abstractmethod + def delete_many(self, item_ids: list[Any]) -> list[T]: + """Delete multiple instances identified by list of IDs ``item_ids``. + + Args: + item_ids: list of Identifiers to be deleted. + + Returns: + The deleted instances. + """ + + @abstractmethod + def exists(self, *filters: FilterTypes, **kwargs: Any) -> bool: + """Return true if the object specified by ``kwargs`` exists. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Identifier of the instance to be retrieved. + + Returns: + True if the instance was found. False if not found. + + """ + + @abstractmethod + def get(self, item_id: Any, **kwargs: Any) -> T: + """Get instance identified by ``item_id``. + + Args: + item_id: Identifier of the instance to be retrieved. + **kwargs: Additional arguments + + Returns: + The retrieved instance. + + Raises: + NotFoundError: If no instance found identified by ``item_id``. + """ + + @abstractmethod + def get_one(self, **kwargs: Any) -> T: + """Get an instance specified by the ``kwargs`` filters if it exists. + + Args: + **kwargs: Instance attribute value filters. + + Returns: + The retrieved instance. + + Raises: + NotFoundError: If no instance found identified by ``kwargs``. + """ + + @abstractmethod + def get_or_create(self, **kwargs: Any) -> tuple[T, bool]: + """Get an instance specified by the ``kwargs`` filters if it exists or create it. + + Args: + **kwargs: Instance attribute value filters. + + Returns: + A tuple that includes the retrieved or created instance, and a boolean on whether the record was created or not + """ + + @abstractmethod + def get_one_or_none(self, **kwargs: Any) -> T | None: + """Get an instance if it exists or None. + + Args: + **kwargs: Instance attribute value filters. + + Returns: + The retrieved instance or None. + """ + + @abstractmethod + def update(self, data: T) -> T: + """Update instance with the attribute values present on ``data``. + + Args: + data: An instance that should have a value for :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` that exists in the + collection. + + Returns: + The updated instance. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + + @abstractmethod + def update_many(self, data: list[T]) -> list[T]: + """Update multiple instances with the attribute values present on instances in ``data``. + + Args: + data: A list of instance that should have a value for :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` that exists in the + collection. + + Returns: + a list of the updated instances. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + + @abstractmethod + def upsert(self, data: T) -> T: + """Update or create instance. + + Updates instance with the attribute values present on ``data``, or creates a new instance if + one doesn't exist. + + Args: + data: Instance to update existing, or be created. Identifier used to determine if an + existing instance exists is the value of an attribute on ``data`` named as value of + :attr:`id_attribute <AbstractAsyncRepository.id_attribute>`. + + Returns: + The updated or created instance. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + + @abstractmethod + def upsert_many(self, data: list[T]) -> list[T]: + """Update or create multiple instances. + + Update instances with the attribute values present on ``data``, or create a new instance if + one doesn't exist. + + Args: + data: Instances to update or created. Identifier used to determine if an + existing instance exists is the value of an attribute on ``data`` named as value of + :attr:`id_attribute <AbstractAsyncRepository.id_attribute>`. + + Returns: + The updated or created instances. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + + @abstractmethod + def list_and_count(self, *filters: FilterTypes, **kwargs: Any) -> tuple[list[T], int]: + """List records with total count. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Instance attribute value filters. + + Returns: + a tuple containing The list of instances, after filtering applied, and a count of records returned by query, ignoring pagination. + """ + + @abstractmethod + def list(self, *filters: FilterTypes, **kwargs: Any) -> list[T]: + """Get a list of instances, optionally filtered. + + Args: + *filters: filters for specific filtering operations + **kwargs: Instance attribute value filters. + + Returns: + The list of instances, after filtering applied + """ + + @abstractmethod + def filter_collection_by_kwargs(self, collection: CollectionT, /, **kwargs: Any) -> CollectionT: + """Filter the collection by kwargs. + + Has ``AND`` semantics where multiple kwargs name/value pairs are provided. + + Args: + collection: the objects to be filtered + **kwargs: key/value pairs such that objects remaining in the collection after filtering + have the property that their attribute named ``key`` has value equal to ``value``. + + + Returns: + The filtered objects + + Raises: + RepositoryError: if a named attribute doesn't exist on :attr:`model_type <AbstractAsyncRepository.model_type>`. + """ + + @staticmethod + def check_not_found(item_or_none: T | None) -> T: + """Raise :class:`NotFoundError` if ``item_or_none`` is ``None``. + + Args: + item_or_none: Item (:class:`T <T>`) to be tested for existence. + + Returns: + The item, if it exists. + """ + if item_or_none is None: + raise NotFoundError("No item found when one was expected") + return item_or_none + + @classmethod + def get_id_attribute_value(cls, item: T | type[T], id_attribute: str | None = None) -> Any: + """Get value of attribute named as :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` on ``item``. + + Args: + item: Anything that should have an attribute named as :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` value. + id_attribute: Allows customization of the unique identifier to use for model fetching. + Defaults to `None`, but can reference any surrogate or candidate key for the table. + + Returns: + The value of attribute on ``item`` named as :attr:`id_attribute <AbstractAsyncRepository.id_attribute>`. + """ + return getattr(item, id_attribute if id_attribute is not None else cls.id_attribute) + + @classmethod + def set_id_attribute_value(cls, item_id: Any, item: T, id_attribute: str | None = None) -> T: + """Return the ``item`` after the ID is set to the appropriate attribute. + + Args: + item_id: Value of ID to be set on instance + item: Anything that should have an attribute named as :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` value. + id_attribute: Allows customization of the unique identifier to use for model fetching. + Defaults to `None`, but can reference any surrogate or candidate key for the table. + + Returns: + Item with ``item_id`` set to :attr:`id_attribute <AbstractAsyncRepository.id_attribute>` + """ + setattr(item, id_attribute if id_attribute is not None else cls.id_attribute, item_id) + return item diff --git a/venv/lib/python3.11/site-packages/litestar/repository/exceptions.py b/venv/lib/python3.11/site-packages/litestar/repository/exceptions.py new file mode 100644 index 0000000..8dad182 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/exceptions.py @@ -0,0 +1,7 @@ +try: + from advanced_alchemy.exceptions import IntegrityError as ConflictError + from advanced_alchemy.exceptions import NotFoundError, RepositoryError +except ImportError: # pragma: no cover + from ._exceptions import ConflictError, NotFoundError, RepositoryError # type: ignore[assignment] + +__all__ = ("ConflictError", "NotFoundError", "RepositoryError") diff --git a/venv/lib/python3.11/site-packages/litestar/repository/filters.py b/venv/lib/python3.11/site-packages/litestar/repository/filters.py new file mode 100644 index 0000000..e0cce48 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/filters.py @@ -0,0 +1,37 @@ +try: + from advanced_alchemy.filters import ( + BeforeAfter, + CollectionFilter, + FilterTypes, + LimitOffset, + NotInCollectionFilter, + NotInSearchFilter, + OnBeforeAfter, + OrderBy, + SearchFilter, + ) +except ImportError: + from ._filters import ( # type: ignore[assignment] + BeforeAfter, + CollectionFilter, + FilterTypes, + LimitOffset, + NotInCollectionFilter, + NotInSearchFilter, + OnBeforeAfter, + OrderBy, + SearchFilter, + ) + + +__all__ = ( + "BeforeAfter", + "CollectionFilter", + "FilterTypes", + "LimitOffset", + "OrderBy", + "SearchFilter", + "NotInCollectionFilter", + "OnBeforeAfter", + "NotInSearchFilter", +) diff --git a/venv/lib/python3.11/site-packages/litestar/repository/handlers.py b/venv/lib/python3.11/site-packages/litestar/repository/handlers.py new file mode 100644 index 0000000..0bc1434 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/handlers.py @@ -0,0 +1,37 @@ +from typing import TYPE_CHECKING + +from litestar.repository.filters import ( + BeforeAfter, + CollectionFilter, + FilterTypes, + LimitOffset, + NotInCollectionFilter, + NotInSearchFilter, + OnBeforeAfter, + OrderBy, + SearchFilter, +) + +if TYPE_CHECKING: + from litestar.config.app import AppConfig + +__all__ = ("signature_namespace_values", "on_app_init") + +signature_namespace_values = { + "BeforeAfter": BeforeAfter, + "OnBeforeAfter": OnBeforeAfter, + "CollectionFilter": CollectionFilter, + "LimitOffset": LimitOffset, + "OrderBy": OrderBy, + "SearchFilter": SearchFilter, + "NotInCollectionFilter": NotInCollectionFilter, + "NotInSearchFilter": NotInSearchFilter, + "FilterTypes": FilterTypes, +} + + +def on_app_init(app_config: "AppConfig") -> "AppConfig": + """Add custom filters for the application during signature modelling.""" + + app_config.signature_namespace.update(signature_namespace_values) + return app_config diff --git a/venv/lib/python3.11/site-packages/litestar/repository/testing/__init__.py b/venv/lib/python3.11/site-packages/litestar/repository/testing/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/testing/__init__.py diff --git a/venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/__init__.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..a008ff2 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/__init__.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/generic_mock_repository.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/generic_mock_repository.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..c871877 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/generic_mock_repository.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/repository/testing/generic_mock_repository.py b/venv/lib/python3.11/site-packages/litestar/repository/testing/generic_mock_repository.py new file mode 100644 index 0000000..5aa094c --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/repository/testing/generic_mock_repository.py @@ -0,0 +1,784 @@ +"""A repository implementation for tests. + +Uses a `dict` for storage. +""" + +from __future__ import annotations + +from datetime import datetime, timezone, tzinfo +from typing import TYPE_CHECKING, Generic, Protocol, TypeVar +from uuid import uuid4 + +from litestar.repository import AbstractAsyncRepository, AbstractSyncRepository, FilterTypes +from litestar.repository.exceptions import ConflictError, RepositoryError + +if TYPE_CHECKING: + from collections.abc import Callable, Hashable, Iterable, MutableMapping + from typing import Any + +__all__ = ("GenericAsyncMockRepository", "GenericSyncMockRepository") + + +class HasID(Protocol): + id: Any + + +ModelT = TypeVar("ModelT", bound="HasID") +AsyncMockRepoT = TypeVar("AsyncMockRepoT", bound="GenericAsyncMockRepository") +SyncMockRepoT = TypeVar("SyncMockRepoT", bound="GenericSyncMockRepository") + + +class GenericAsyncMockRepository(AbstractAsyncRepository[ModelT], Generic[ModelT]): + """A repository implementation for tests. + + Uses a :class:`dict` for storage. + """ + + collection: MutableMapping[Hashable, ModelT] + model_type: type[ModelT] + match_fields: list[str] | str | None = None + + _model_has_created_at: bool + _model_has_updated_at: bool + + def __init__( + self, id_factory: Callable[[], Any] = uuid4, tz: tzinfo = timezone.utc, allow_ids_on_add: bool = False, **_: Any + ) -> None: + super().__init__() + self._id_factory = id_factory + self.tz = tz + self.allow_ids_on_add = allow_ids_on_add + + @classmethod + def __class_getitem__(cls: type[AsyncMockRepoT], item: type[ModelT]) -> type[AsyncMockRepoT]: + """Add collection to ``_collections`` for the type. + + Args: + item: The type that the class has been parametrized with. + """ + return type( # pyright:ignore + f"{cls.__name__}[{item.__name__}]", + (cls,), + { + "collection": {}, + "model_type": item, + "_model_has_created_at": hasattr(item, "created_at"), + "_model_has_updated_at": hasattr(item, "updated_at"), + }, + ) + + def _find_or_raise_not_found(self, item_id: Any) -> ModelT: + return self.check_not_found(self.collection.get(item_id)) + + def _find_or_none(self, item_id: Any) -> ModelT | None: + return self.collection.get(item_id) + + def _now(self) -> datetime: + return datetime.now(tz=self.tz).replace(tzinfo=None) + + def _update_audit_attributes(self, data: ModelT, now: datetime | None = None, do_created: bool = False) -> ModelT: + now = now or self._now() + if self._model_has_updated_at: + data.updated_at = now # type:ignore[attr-defined] + if do_created: + data.created_at = now # type:ignore[attr-defined] + return data + + async def add(self, data: ModelT) -> ModelT: + """Add ``data`` to the collection. + + Args: + data: Instance to be added to the collection. + + Returns: + The added instance. + """ + if self.allow_ids_on_add is False and self.get_id_attribute_value(data) is not None: + raise ConflictError("`add()` received identified item.") + self._update_audit_attributes(data, do_created=True) + if self.allow_ids_on_add is False: + id_ = self._id_factory() + self.set_id_attribute_value(id_, data) + self.collection[data.id] = data + return data + + async def add_many(self, data: Iterable[ModelT]) -> list[ModelT]: + """Add multiple ``data`` to the collection. + + Args: + data: Instance to be added to the collection. + + Returns: + The added instance. + """ + now = self._now() + for data_row in data: + if self.allow_ids_on_add is False and self.get_id_attribute_value(data_row) is not None: + raise ConflictError("`add()` received identified item.") + + self._update_audit_attributes(data_row, do_created=True, now=now) + if self.allow_ids_on_add is False: + id_ = self._id_factory() + self.set_id_attribute_value(id_, data_row) + self.collection[data_row.id] = data_row + return list(data) + + async def delete(self, item_id: Any) -> ModelT: + """Delete instance identified by ``item_id``. + + Args: + item_id: Identifier of instance to be deleted. + + Returns: + The deleted instance. + + Raises: + NotFoundError: If no instance found identified by ``item_id``. + """ + try: + return self._find_or_raise_not_found(item_id) + finally: + del self.collection[item_id] + + async def delete_many(self, item_ids: list[Any]) -> list[ModelT]: + """Delete instances identified by list of identifiers ``item_ids``. + + Args: + item_ids: list of identifiers of instances to be deleted. + + Returns: + The deleted instances. + + """ + instances: list[ModelT] = [] + for item_id in item_ids: + obj = await self.get_one_or_none(**{self.id_attribute: item_id}) + if obj: + obj = await self.delete(obj.id) + instances.append(obj) + return instances + + async def exists(self, *filters: FilterTypes, **kwargs: Any) -> bool: + """Return true if the object specified by ``kwargs`` exists. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Identifier of the instance to be retrieved. + + Returns: + True if the instance was found. False if not found.. + + """ + existing = await self.count(*filters, **kwargs) + return bool(existing) + + async def get(self, item_id: Any, **kwargs: Any) -> ModelT: + """Get instance identified by ``item_id``. + + Args: + item_id: Identifier of the instance to be retrieved. + **kwargs: additional arguments + + Returns: + The retrieved instance. + + Raises: + NotFoundError: If no instance found identified by ``item_id``. + """ + return self._find_or_raise_not_found(item_id) + + async def get_or_create(self, match_fields: list[str] | str | None = None, **kwargs: Any) -> tuple[ModelT, bool]: + """Get instance identified by ``kwargs`` or create if it doesn't exist. + + Args: + match_fields: a list of keys to use to match the existing model. When empty, all fields are matched. + **kwargs: Identifier of the instance to be retrieved. + + Returns: + a tuple that includes the instance and whether it needed to be created. + + """ + match_fields = match_fields or self.match_fields + if isinstance(match_fields, str): + match_fields = [match_fields] + if match_fields: + match_filter = { + field_name: field_value + for field_name in match_fields + if (field_value := kwargs.get(field_name)) is not None + } + else: + match_filter = kwargs + existing = await self.get_one_or_none(**match_filter) + if existing: + for field_name, new_field_value in kwargs.items(): + field = getattr(existing, field_name, None) + if field and field != new_field_value: + setattr(existing, field_name, new_field_value) + + return existing, False + return await self.add(self.model_type(**kwargs)), True # pyright: ignore[reportGeneralTypeIssues] + + async def get_one(self, **kwargs: Any) -> ModelT: + """Get instance identified by query filters. + + Args: + **kwargs: Instance attribute value filters. + + Returns: + The retrieved instance or None + + Raises: + NotFoundError: If no instance found identified by ``kwargs``. + """ + data = await self.list(**kwargs) + return self.check_not_found(data[0] if data else None) + + async def get_one_or_none(self, **kwargs: Any) -> ModelT | None: + """Get instance identified by query filters or None if not found. + + Args: + **kwargs: Instance attribute value filters. + + Returns: + The retrieved instance or None + """ + data = await self.list(**kwargs) + return data[0] if data else None + + async def count(self, *filters: FilterTypes, **kwargs: Any) -> int: + """Count of rows returned by query. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Instance attribute value filters. + + Returns: + Count of instances in collection, ignoring pagination. + """ + return len(await self.list(*filters, **kwargs)) + + async def update(self, data: ModelT) -> ModelT: + """Update instance with the attribute values present on ``data``. + + Args: + data: An instance that should have a value for :attr:`id_attribute <AsyncGenericMockRepository.id_attribute>` that exists in the + collection. + + Returns: + The updated instance. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + item = self._find_or_raise_not_found(self.get_id_attribute_value(data)) + self._update_audit_attributes(data, do_created=False) + for key, val in model_items(data): + setattr(item, key, val) + return item + + async def update_many(self, data: list[ModelT]) -> list[ModelT]: + """Update instances with the attribute values present on ``data``. + + Args: + data: A list of instances that should have a value for :attr:`id_attribute <AsyncGenericMockRepository.id_attribute>` + that exists in the collection. + + Returns: + The updated instances. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + items = [self._find_or_raise_not_found(self.get_id_attribute_value(row)) for row in data] + now = self._now() + for item in items: + self._update_audit_attributes(item, do_created=False, now=now) + for key, val in model_items(item): + setattr(item, key, val) + return items + + async def upsert(self, data: ModelT) -> ModelT: + """Update or create instance. + + Updates instance with the attribute values present on ``data``, or creates a new instance if + one doesn't exist. + + Args: + data: Instance to update existing, or be created. Identifier used to determine if an + existing instance exists is the value of an attribute on `data` named as value of + :attr:`id_attribute <AsyncGenericMockRepository.id_attribute>`. + + Returns: + The updated or created instance. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + item_id = self.get_id_attribute_value(data) + if item_id in self.collection: + return await self.update(data) + return await self.add(data) + + async def upsert_many(self, data: list[ModelT]) -> list[ModelT]: + """Update or create multiple instance. + + Update instance with the attribute values present on ``data``, or create a new instance if + one doesn't exist. + + Args: + data: List of instances to update existing, or be created. Identifier used to determine if an + existing instance exists is the value of an attribute on `data` named as value of + :attr:`id_attribute <AsyncGenericMockRepository.id_attribute>`. + + Returns: + The updated or created instances. + """ + data_to_update = [row for row in data if self._find_or_none(self.get_id_attribute_value(row)) is not None] + data_to_add = [row for row in data if self._find_or_none(self.get_id_attribute_value(row)) is None] + + updated_items = await self.update_many(data_to_update) + added_items = await self.add_many(data_to_add) + return updated_items + added_items + + async def list_and_count( + self, + *filters: FilterTypes, + **kwargs: Any, + ) -> tuple[list[ModelT], int]: + """Get a list of instances, optionally filtered with a total row count. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Instance attribute value filters. + + Returns: + List of instances, and count of records returned by query, ignoring pagination. + """ + return await self.list(*filters, **kwargs), await self.count(*filters, **kwargs) + + async def list(self, *filters: FilterTypes, **kwargs: Any) -> list[ModelT]: + """Get a list of instances, optionally filtered. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Instance attribute value filters. + + Returns: + The list of instances, after filtering applied. + """ + return list(self.filter_collection_by_kwargs(self.collection, **kwargs).values()) + + def filter_collection_by_kwargs( # type:ignore[override] + self, collection: MutableMapping[Hashable, ModelT], /, **kwargs: Any + ) -> MutableMapping[Hashable, ModelT]: + """Filter the collection by kwargs. + + Args: + collection: set of objects to filter + **kwargs: key/value pairs such that objects remaining in the collection after filtering + have the property that their attribute named ``key`` has value equal to ``value``. + """ + new_collection: dict[Hashable, ModelT] = {} + for item in self.collection.values(): + try: + if all(getattr(item, name) == value for name, value in kwargs.items()): + new_collection[item.id] = item + except AttributeError as orig: + raise RepositoryError from orig + return new_collection + + @classmethod + def seed_collection(cls, instances: Iterable[ModelT]) -> None: + """Seed the collection for repository type. + + Args: + instances: the instances to be added to the collection. + """ + for instance in instances: + cls.collection[cls.get_id_attribute_value(instance)] = instance + + @classmethod + def clear_collection(cls) -> None: + """Empty the collection for repository type.""" + cls.collection = {} + + +class GenericSyncMockRepository(AbstractSyncRepository[ModelT], Generic[ModelT]): + """A repository implementation for tests. + + Uses a :class:`dict` for storage. + """ + + collection: MutableMapping[Hashable, ModelT] + model_type: type[ModelT] + match_fields: list[str] | str | None = None + + _model_has_created_at: bool + _model_has_updated_at: bool + + def __init__( + self, + id_factory: Callable[[], Any] = uuid4, + tz: tzinfo = timezone.utc, + allow_ids_on_add: bool = False, + **_: Any, + ) -> None: + super().__init__() + self._id_factory = id_factory + self.tz = tz + self.allow_ids_on_add = allow_ids_on_add + + @classmethod + def __class_getitem__(cls: type[SyncMockRepoT], item: type[ModelT]) -> type[SyncMockRepoT]: + """Add collection to ``_collections`` for the type. + + Args: + item: The type that the class has been parametrized with. + """ + return type( # pyright:ignore + f"{cls.__name__}[{item.__name__}]", + (cls,), + { + "collection": {}, + "model_type": item, + "_model_has_created_at": hasattr(item, "created_at"), + "_model_has_updated_at": hasattr(item, "updated_at"), + }, + ) + + def _find_or_raise_not_found(self, item_id: Any) -> ModelT: + return self.check_not_found(self.collection.get(item_id)) + + def _find_or_none(self, item_id: Any) -> ModelT | None: + return self.collection.get(item_id) + + def _now(self) -> datetime: + return datetime.now(tz=self.tz).replace(tzinfo=None) + + def _update_audit_attributes(self, data: ModelT, now: datetime | None = None, do_created: bool = False) -> ModelT: + now = now or self._now() + if self._model_has_updated_at: + data.updated_at = now # type:ignore[attr-defined] + if do_created: + data.created_at = now # type:ignore[attr-defined] + return data + + def add(self, data: ModelT) -> ModelT: + """Add ``data`` to the collection. + + Args: + data: Instance to be added to the collection. + + Returns: + The added instance. + """ + if self.allow_ids_on_add is False and self.get_id_attribute_value(data) is not None: + raise ConflictError("`add()` received identified item.") + self._update_audit_attributes(data, do_created=True) + if self.allow_ids_on_add is False: + id_ = self._id_factory() + self.set_id_attribute_value(id_, data) + self.collection[data.id] = data + return data + + def add_many(self, data: Iterable[ModelT]) -> list[ModelT]: + """Add multiple ``data`` to the collection. + + Args: + data: Instance to be added to the collection. + + Returns: + The added instance. + """ + now = self._now() + for data_row in data: + if self.allow_ids_on_add is False and self.get_id_attribute_value(data_row) is not None: + raise ConflictError("`add()` received identified item.") + + self._update_audit_attributes(data_row, do_created=True, now=now) + if self.allow_ids_on_add is False: + id_ = self._id_factory() + self.set_id_attribute_value(id_, data_row) + self.collection[data_row.id] = data_row + return list(data) + + def delete(self, item_id: Any) -> ModelT: + """Delete instance identified by ``item_id``. + + Args: + item_id: Identifier of instance to be deleted. + + Returns: + The deleted instance. + + Raises: + NotFoundError: If no instance found identified by ``item_id``. + """ + try: + return self._find_or_raise_not_found(item_id) + finally: + del self.collection[item_id] + + def delete_many(self, item_ids: list[Any]) -> list[ModelT]: + """Delete instances identified by list of identifiers ``item_ids``. + + Args: + item_ids: list of identifiers of instances to be deleted. + + Returns: + The deleted instances. + + """ + instances: list[ModelT] = [] + for item_id in item_ids: + if obj := self.get_one_or_none(**{self.id_attribute: item_id}): + obj = self.delete(obj.id) + instances.append(obj) + return instances + + def exists(self, *filters: FilterTypes, **kwargs: Any) -> bool: + """Return true if the object specified by ``kwargs`` exists. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Identifier of the instance to be retrieved. + + Returns: + True if the instance was found. False if not found.. + + """ + existing = self.count(*filters, **kwargs) + return bool(existing) + + def get(self, item_id: Any, **kwargs: Any) -> ModelT: + """Get instance identified by ``item_id``. + + Args: + item_id: Identifier of the instance to be retrieved. + **kwargs: additional arguments + + Returns: + The retrieved instance. + + Raises: + NotFoundError: If no instance found identified by ``item_id``. + """ + return self._find_or_raise_not_found(item_id) + + def get_or_create(self, match_fields: list[str] | str | None = None, **kwargs: Any) -> tuple[ModelT, bool]: + """Get instance identified by ``kwargs`` or create if it doesn't exist. + + Args: + match_fields: a list of keys to use to match the existing model. When empty, all fields are matched. + **kwargs: Identifier of the instance to be retrieved. + + Returns: + a tuple that includes the instance and whether it needed to be created. + + """ + match_fields = match_fields or self.match_fields + if isinstance(match_fields, str): + match_fields = [match_fields] + if match_fields: + match_filter = { + field_name: field_value + for field_name in match_fields + if (field_value := kwargs.get(field_name)) is not None + } + else: + match_filter = kwargs + if existing := self.get_one_or_none(**match_filter): + for field_name, new_field_value in kwargs.items(): + field = getattr(existing, field_name, None) + if field and field != new_field_value: + setattr(existing, field_name, new_field_value) + + return existing, False + return self.add(self.model_type(**kwargs)), True # pyright: ignore[reportGeneralTypeIssues] + + def get_one(self, **kwargs: Any) -> ModelT: + """Get instance identified by query filters. + + Args: + **kwargs: Instance attribute value filters. + + Returns: + The retrieved instance or None + + Raises: + NotFoundError: If no instance found identified by ``kwargs``. + """ + data = self.list(**kwargs) + return self.check_not_found(data[0] if data else None) + + def get_one_or_none(self, **kwargs: Any) -> ModelT | None: + """Get instance identified by query filters or None if not found. + + Args: + **kwargs: Instance attribute value filters. + + Returns: + The retrieved instance or None + """ + data = self.list(**kwargs) + return data[0] if data else None + + def count(self, *filters: FilterTypes, **kwargs: Any) -> int: + """Count of rows returned by query. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Instance attribute value filters. + + Returns: + Count of instances in collection, ignoring pagination. + """ + return len(self.list(*filters, **kwargs)) + + def update(self, data: ModelT) -> ModelT: + """Update instance with the attribute values present on ``data``. + + Args: + data: An instance that should have a value for :attr:`id_attribute <AsyncGenericMockRepository.id_attribute>` that exists in the + collection. + + Returns: + The updated instance. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + item = self._find_or_raise_not_found(self.get_id_attribute_value(data)) + self._update_audit_attributes(data, do_created=False) + for key, val in model_items(data): + setattr(item, key, val) + return item + + def update_many(self, data: list[ModelT]) -> list[ModelT]: + """Update instances with the attribute values present on ``data``. + + Args: + data: A list of instances that should have a value for :attr:`id_attribute <AsyncGenericMockRepository.id_attribute>` + that exists in the collection. + + Returns: + The updated instances. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + items = [self._find_or_raise_not_found(self.get_id_attribute_value(row)) for row in data] + now = self._now() + for item in items: + self._update_audit_attributes(item, do_created=False, now=now) + for key, val in model_items(item): + setattr(item, key, val) + return items + + def upsert(self, data: ModelT) -> ModelT: + """Update or create instance. + + Updates instance with the attribute values present on ``data``, or creates a new instance if + one doesn't exist. + + Args: + data: Instance to update existing, or be created. Identifier used to determine if an + existing instance exists is the value of an attribute on `data` named as value of + :attr:`id_attribute <AsyncGenericMockRepository.id_attribute>`. + + Returns: + The updated or created instance. + + Raises: + NotFoundError: If no instance found with same identifier as ``data``. + """ + item_id = self.get_id_attribute_value(data) + return self.update(data) if item_id in self.collection else self.add(data) + + def upsert_many(self, data: list[ModelT]) -> list[ModelT]: + """Update or create multiple instance. + + Update instance with the attribute values present on ``data``, or create a new instance if + one doesn't exist. + + Args: + data: List of instances to update existing, or be created. Identifier used to determine if an + existing instance exists is the value of an attribute on `data` named as value of + :attr:`id_attribute <AsyncGenericMockRepository.id_attribute>`. + + Returns: + The updated or created instances. + """ + data_to_update = [row for row in data if self._find_or_none(self.get_id_attribute_value(row)) is not None] + data_to_add = [row for row in data if self._find_or_none(self.get_id_attribute_value(row)) is None] + + updated_items = self.update_many(data_to_update) + added_items = self.add_many(data_to_add) + return updated_items + added_items + + def list_and_count( + self, + *filters: FilterTypes, + **kwargs: Any, + ) -> tuple[list[ModelT], int]: + """Get a list of instances, optionally filtered with a total row count. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Instance attribute value filters. + + Returns: + List of instances, and count of records returned by query, ignoring pagination. + """ + return self.list(*filters, **kwargs), self.count(*filters, **kwargs) + + def list(self, *filters: FilterTypes, **kwargs: Any) -> list[ModelT]: + """Get a list of instances, optionally filtered. + + Args: + *filters: Types for specific filtering operations. + **kwargs: Instance attribute value filters. + + Returns: + The list of instances, after filtering applied. + """ + return list(self.filter_collection_by_kwargs(self.collection, **kwargs).values()) + + def filter_collection_by_kwargs( # type:ignore[override] + self, collection: MutableMapping[Hashable, ModelT], /, **kwargs: Any + ) -> MutableMapping[Hashable, ModelT]: + """Filter the collection by kwargs. + + Args: + collection: set of objects to filter + **kwargs: key/value pairs such that objects remaining in the collection after filtering + have the property that their attribute named ``key`` has value equal to ``value``. + """ + new_collection: dict[Hashable, ModelT] = {} + for item in self.collection.values(): + try: + if all(getattr(item, name) == value for name, value in kwargs.items()): + new_collection[item.id] = item + except AttributeError as orig: + raise RepositoryError from orig + return new_collection + + @classmethod + def seed_collection(cls, instances: Iterable[ModelT]) -> None: + """Seed the collection for repository type. + + Args: + instances: the instances to be added to the collection. + """ + for instance in instances: + cls.collection[cls.get_id_attribute_value(instance)] = instance + + @classmethod + def clear_collection(cls) -> None: + """Empty the collection for repository type.""" + cls.collection = {} + + +def model_items(model: Any) -> list[tuple[str, Any]]: + return [(k, v) for k, v in model.__dict__.items() if not k.startswith("_")] |