summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/repository
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
commit12cf076118570eebbff08c6b3090e0d4798447a1 (patch)
tree3ba25e17e3c3a5e82316558ba3864b955919ff72 /venv/lib/python3.11/site-packages/litestar/repository
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/repository')
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/__init__.py14
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/__pycache__/__init__.cpython-311.pycbin609 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_exceptions.cpython-311.pycbin1069 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_filters.cpython-311.pycbin4350 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/__pycache__/exceptions.cpython-311.pycbin585 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/__pycache__/filters.cpython-311.pycbin898 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/__pycache__/handlers.cpython-311.pycbin1250 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/_exceptions.py15
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/_filters.py117
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/abc/__init__.py7
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/__init__.cpython-311.pycbin373 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_async.cpython-311.pycbin13618 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_sync.cpython-311.pycbin13523 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/abc/_async.py303
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/abc/_sync.py305
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/exceptions.py7
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/filters.py37
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/handlers.py37
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/testing/__init__.py0
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/__init__.cpython-311.pycbin211 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/generic_mock_repository.cpython-311.pycbin32712 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/repository/testing/generic_mock_repository.py784
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
deleted file mode 100644
index 42bf9be..0000000
--- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/__init__.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 0f48627..0000000
--- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_exceptions.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 2f16cb4..0000000
--- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/_filters.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 3c9f01e..0000000
--- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/exceptions.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 87d90db..0000000
--- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/filters.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index abe227b..0000000
--- a/venv/lib/python3.11/site-packages/litestar/repository/__pycache__/handlers.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 710a74b..0000000
--- a/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/__init__.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 24217cc..0000000
--- a/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_async.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 7a69407..0000000
--- a/venv/lib/python3.11/site-packages/litestar/repository/abc/__pycache__/_sync.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index a008ff2..0000000
--- a/venv/lib/python3.11/site-packages/litestar/repository/testing/__pycache__/__init__.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted 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
Binary files differ
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("_")]