diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/repository')
22 files changed, 0 insertions, 1626 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 deleted file mode 100644 index 62e3f83..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/__init__.py +++ /dev/null @@ -1,14 +0,0 @@ -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 differdeleted file mode 100644 index 42bf9be..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/__init__.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 0f48627..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_exceptions.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 2f16cb4..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_filters.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 3c9f01e..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/exceptions.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 87d90db..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/filters.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index abe227b..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/handlers.cpython-311.pyc +++ /dev/null diff --git a/venv/lib/python3.11/site-packages/litestar/repository/_exceptions.py b/venv/lib/python3.11/site-packages/litestar/repository/_exceptions.py deleted file mode 100644 index 1c2b7be..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/_exceptions.py +++ /dev/null @@ -1,15 +0,0 @@ -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 deleted file mode 100644 index f6b787e..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/_filters.py +++ /dev/null @@ -1,117 +0,0 @@ -"""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 deleted file mode 100644 index def6bb4..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/abc/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -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 differdeleted file mode 100644 index 710a74b..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/__init__.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 24217cc..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_async.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 7a69407..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_sync.cpython-311.pyc +++ /dev/null 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 deleted file mode 100644 index 85ca139..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/abc/_async.py +++ /dev/null @@ -1,303 +0,0 @@ -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 deleted file mode 100644 index d667fc2..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/abc/_sync.py +++ /dev/null @@ -1,305 +0,0 @@ -# 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 deleted file mode 100644 index 8dad182..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/exceptions.py +++ /dev/null @@ -1,7 +0,0 @@ -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 deleted file mode 100644 index e0cce48..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/filters.py +++ /dev/null @@ -1,37 +0,0 @@ -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 deleted file mode 100644 index 0bc1434..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/handlers.py +++ /dev/null @@ -1,37 +0,0 @@ -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 deleted file mode 100644 index e69de29..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/testing/__init__.py +++ /dev/null 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 differdeleted file mode 100644 index a008ff2..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/__init__.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index c871877..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/generic_mock_repository.cpython-311.pyc +++ /dev/null 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 deleted file mode 100644 index 5aa094c..0000000 --- a/venv/lib/python3.11/site-packages/litestar/repository/testing/generic_mock_repository.py +++ /dev/null @@ -1,784 +0,0 @@ -"""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("_")] |