# AbstractRepo - Python Repository Pattern Implementation
[](https://pypi.org/project/abstractrepo/)
[](https://github.com/Smoren/abstractrepo-pypi/releases)
[](https://coveralls.io/github/Smoren/abstractrepo-pypi?branch=master)
[](https://pepy.tech/projects/abstractrepo)
[](https://github.com/Smoren/abstractrepo-pypi/actions)
[](https://github.com/Smoren/abstractrepo-pypi/blob/master/LICENSE)
The `AbstractRepo` library provides a robust and flexible abstraction layer for interacting with various data storage systems in Python. It implements the widely recognized Repository Pattern, offering a clean and consistent API for common data operations such as Create, Read, Update, and Delete (CRUD). This design promotes a clear separation of concerns, making your application logic independent of the underlying data persistence mechanism. This allows for easy switching between different databases or storage solutions without significant changes to your core business logic.
## Key Features
* **CRUD Operations:** Comprehensive support for standard data manipulation operations.
* **Specifications Pattern:** A powerful and flexible mechanism for defining complex query criteria based on business rules, enabling highly customizable data retrieval.
* **Ordering Options:** Advanced sorting capabilities, including control over the placement of `NULL` values.
* **Pagination Support:** Efficient handling of large datasets through limit and offset-based pagination.
* **Strong Typing:** Leverages Python's type hinting for improved code readability, maintainability, and early error detection.
* **Extensibility:** Designed with extensibility in mind, allowing for easy integration with various database technologies and custom data sources.
* **In-Memory Implementation:** Includes a built-in list-based repository implementation, ideal for testing, development, and rapid prototyping.
* **Asynchronous Support:** Provides interfaces and base implementations for asynchronous repository operations, crucial for modern, high-performance applications.
## Implementations
* [SQLAlchemy implementation](https://github.com/Smoren/abstractrepo-sqlalchemy-pypi)
* [List-based in-memory implementation](#list-based-implementation-listbasedcrudrepository)
## Installation
To get started with `AbstractRepo`, install it using pip:
```shell
pip install abstractrepo
```
## Table of Contents
* [Core Components and Usage](#core-components-and-usage)
* [Repository Interface (`CrudRepositoryInterface`)](#repository-interface-crudrepositoryinterface)
* [List-Based Implementation (`ListBasedCrudRepository`)](#list-based-implementation-listbasedcrudrepository)
* [Asynchronous Repositories (`AsyncCrudRepositoryInterface`)](#asynchronous-repositories)
* [Specifications](#specifications)
* [Ordering](#ordering)
* [Pagination](#pagination)
* [Exception Handling](#exception-handling)
* [Examples](#examples)
* [Complete Synchronous Example](#complete-synchronous-example)
* [Complete Asynchronous Example](#complete-asynchronous-example)
* [API Reference](#api-reference)
* [Repository Methods](#repository-methods)
* [Specification Types](#specification-types)
* [Ordering Options](#ordering-options)
* [Pagination Options](#pagination-options)
* [Best Practices](#best-practices)
* [Dependencies](#dependencies)
* [License](#license)
## Core Components and Usage
### Repository Interface (`CrudRepositoryInterface`)
The `CrudRepositoryInterface` defines the contract for all synchronous repositories. It specifies the standard CRUD operations and other essential methods that any concrete repository implementation must adhere to.
```python
import abc
from pydantic import BaseModel
from abstractrepo.repo import CrudRepositoryInterface
class User(BaseModel):
id: int
username: str
password: str
display_name: str
class UserCreateForm(BaseModel):
username: str
password: str
display_name: str
class UserUpdateForm(BaseModel):
display_name: str
class UserRepositoryInterface(CrudRepositoryInterface[User, int, UserCreateForm, UserUpdateForm], abc.ABC):
pass
class UserRepository(UserRepositoryInterface):
# Implement abstract methods here
...
```
**Key Methods:**
| Method | Parameters | Returns | Description |
|:-----------------|:-------------------------------------------------|:---------------|:-----------------------------------------------------------------------------------------------|
| `get_collection` | `filter_spec`, `order_options`, `paging_options` | `List[TModel]` | Retrieves a collection of items based on filtering, sorting, and pagination options. |
| `count` | `filter_spec` | `int` | Returns the total count of items matching the given filter specification. |
| `get_item` | `item_id` | `TModel` | Retrieves a single item by its unique identifier. Raises `ItemNotFoundException` if not found. |
| `exists` | `item_id` | `bool` | Checks if an item with the specified ID exists in the repository. |
| `create` | `form` | `TModel` | Creates a new item in the repository using the provided creation form. |
| `update` | `item_id`, `form` | `TModel` | Updates an existing item identified by its ID with data from the update form. |
| `delete` | `item_id` | `TModel` | Deletes an item from the repository by its ID. |
| `model_class` | (Property) | `Type[TModel]` | Returns the model class associated with the repository. |
### List-Based Implementation (`ListBasedCrudRepository`)
The `ListBasedCrudRepository` provides a concrete, in-memory implementation of the `CrudRepositoryInterface`. It's particularly useful for testing, development, and scenarios where a simple, non-persistent data store is sufficient.
```python
import abc
from typing import Optional, List, Type
from pydantic import BaseModel
from abstractrepo.repo import CrudRepositoryInterface, ListBasedCrudRepository
from abstractrepo.specification import SpecificationInterface, AttributeSpecification, Operator
from abstractrepo.exceptions import ItemNotFoundException, UniqueViolationException
class User(BaseModel):
id: int
username: str
password: str
display_name: str
class UserCreateForm(BaseModel):
username: str
password: str
display_name: str
class UserUpdateForm(BaseModel):
display_name: str
class UserRepositoryInterface(CrudRepositoryInterface[User, int, UserCreateForm, UserUpdateForm], abc.ABC):
pass
class ListBasedUserRepository(
ListBasedCrudRepository[User, int, UserCreateForm, UserUpdateForm],
UserRepositoryInterface,
):
_next_id: int
def __init__(self, items: Optional[List[User]] = None):
super().__init__(items)
self._next_id = 0
def get_by_username(self, username: str) -> User:
items = self.get_collection(AttributeSpecification("username", username))
if len(items) == 0:
raise ItemNotFoundException(User)
return items[0]
@property
def model_class(self) -> Type[User]:
return User
def _create_model(self, form: UserCreateForm, new_id: int) -> User:
if self._username_exists(form.username):
raise UniqueViolationException(User, "create", form)
return User(
id=new_id,
username=form.username,
password=form.password,
display_name=form.display_name,
)
def _update_model(self, model: User, form: UserUpdateForm) -> User:
model.display_name = form.display_name
return model
def _username_exists(self, username: str) -> bool:
return self.count(AttributeSpecification("username", username)) > 0
def _generate_id(self) -> int:
self._next_id += 1
return self._next_id
def _get_id_filter_specification(self, item_id: int) -> SpecificationInterface[User, bool]:
return AttributeSpecification("id", item_id, Operator.E)
```
### Asynchronous Repositories
`AbstractRepo` provides full support for asynchronous operations, allowing you to build non-blocking data access layers for high-performance applications. The `AsyncCrudRepositoryInterface` defines the asynchronous contract, and `AsyncListBasedCrudRepository` offers an in-memory asynchronous implementation.
This interface mirrors the synchronous `CrudRepositoryInterface` but with `async` methods, enabling seamless integration with `asyncio` and other asynchronous frameworks.
```python
import abc
from typing import TypeVar
from pydantic import BaseModel
from abstractrepo.repo import AsyncCrudRepositoryInterface
class User(BaseModel):
id: int
username: str
password: str
display_name: str
class UserCreateForm(BaseModel):
username: str
password: str
display_name: str
class UserUpdateForm(BaseModel):
display_name: str
class AsyncUserRepositoryInterface(AsyncCrudRepositoryInterface[User, int, UserCreateForm, UserUpdateForm], abc.ABC):
pass
class AsyncUserRepository(AsyncUserRepositoryInterface):
# Implement abstract async methods here
...
```
### Specifications
The Specifications Pattern allows you to define flexible and reusable filtering logic. You can combine simple specifications to build complex queries.
**Filtering with Specifications:**
```python
from abstractrepo.specification import AttributeSpecification, AndSpecification, OrSpecification, Operator
# Single attribute filter
active_users_spec = AttributeSpecification("is_active", True)
# Complex filter combining AND and OR operations
premium_filter_spec = AndSpecification(
AttributeSpecification("plan", "premium"),
OrSpecification(
AttributeSpecification("age", 30, Operator.GTE),
AttributeSpecification("join_date", "2023-01-01", Operator.GT)
)
)
```
**Supported Operators:**
`AbstractRepo` provides a rich set of operators for various comparison and matching needs:
| Operator | Description |
| :------- |:-------------------------------|
| `E` | Equal |
| `NE` | Not Equal |
| `GT` | Greater Than |
| `LT` | Less Than |
| `GTE` | Greater Than or Equal |
| `LTE` | Less Than or Equal |
| `LIKE` | Case-Sensitive Pattern Match |
| `ILIKE` | Case-Insensitive Pattern Match |
| `IN` | In List |
| `NOT_IN` | Not In List |
### Ordering
Control the order of retrieved items using `OrderOptions` and `OrderOption`. You can specify the attribute to sort by, the direction (ascending or descending), and how `None` values should be handled.
```python
from abstractrepo.order import OrderOptionsBuilder, OrderOptions, OrderOption, OrderDirection, NonesOrder
# Single field ordering
ordering_by_name = OrderOptions(
OrderOption("name", OrderDirection.ASC, NonesOrder.LAST)
)
# Multi-field ordering using OrderOptionsBuilder for chaining
complex_ordering = OrderOptionsBuilder() \
.add("priority", OrderDirection.DESC) \
.add("created_at", OrderDirection.ASC, NonesOrder.LAST) \
.build()
```
### Pagination
Manage large result sets efficiently with `PagingOptions` for limit/offset-based pagination and `PageResolver` for page-number-based navigation.
```python
from abstractrepo.paging import PagingOptions, PageResolver
# Manual paging with limit and offset
manual_paging = PagingOptions(limit=10, offset=20)
# Page-based resolver for consistent page navigation
resolver = PageResolver(page_size=25)
page_3_options = resolver.get_page(3) # Retrieves PagingOptions for the 3rd page
```
### Exception Handling
`AbstractRepo` defines specific exceptions to handle common repository-related errors, allowing for robust error management in your application.
```python
from abstractrepo.exceptions import (
ItemNotFoundException,
UniqueViolationException,
RelationViolationException,
)
try:
repo.get_item(999) # Attempt to retrieve a non-existent item
except ItemNotFoundException as e:
print(f"Error: {e}") # Handle the case where the item is not found
```
## Examples
### Complete Synchronous Example
This example demonstrates the full lifecycle of a synchronous repository using the `ListBasedCrudRepository`.
```python
from abstractrepo.repo import ListBasedCrudRepository
from abstractrepo.specification import AttributeSpecification, Operator
from abstractrepo.order import OrderOptions, OrderOption, OrderDirection
from abstractrepo.paging import PagingOptions
from pydantic import BaseModel
# Define your data model
class User(BaseModel):
id: int
name: str
email: str
# Define forms for creation and update (can be pure classes Pydantic models)
class UserCreateForm(BaseModel):
name: str
email: str
class UserUpdateForm(BaseModel):
name: str | None = None
email: str | None = None
# Implement your concrete repository
class ConcreteUserRepository(ListBasedCrudRepository[User, int, UserCreateForm, UserUpdateForm]):
_next_id: int = 0
def __init__(self, items: list[User] | None = None):
super().__init__(items)
if items:
self._next_id = max(item.id for item in items) + 1
@property
def model_class(self) -> type[User]:
return User
def _create_model(self, form: UserCreateForm, new_id: int) -> User:
# In a real scenario, you might check for unique constraints here
return User(id=new_id, name=form.name, email=form.email)
def _update_model(self, model: User, form: UserUpdateForm) -> User:
if form.name is not None:
model.name = form.name
if form.email is not None:
model.email = form.email
return model
def _generate_id(self) -> int:
self._next_id += 1
return self._next_id
def _get_id_filter_specification(self, item_id: int) -> AttributeSpecification[User, bool]:
return AttributeSpecification("id", item_id, Operator.E)
# --- Usage Example ---
# Initialize repository
user_repo = ConcreteUserRepository()
# Create users
alice = user_repo.create(UserCreateForm(name="Alice", email="alice@example.com"))
bob = user_repo.create(UserCreateForm(name="Bob", email="bob@example.com"))
charlie = user_repo.create(UserCreateForm(name="Charlie", email="charlie@example.com"))
print(f"Created users: {alice.name}, {bob.name}, {charlie.name}")
# Query with specifications
b_users = user_repo.get_collection(
filter_spec=AttributeSpecification("name", "B%", Operator.LIKE),
order_options=OrderOptions(OrderOption("email", OrderDirection.ASC)),
paging_options=PagingOptions(limit=5)
)
print(f"Users with name starting with 'B': {[u.name for u in b_users]}")
# Get a single user by ID
retrieved_alice = user_repo.get_item(alice.id)
print(f"Retrieved user by ID: {retrieved_alice.name}")
# Update a user
user_repo.update(bob.id, UserUpdateForm(email="robert@example.com"))
updated_bob = user_repo.get_item(bob.id)
print(f"Updated Bob's email to: {updated_bob.email}")
# Get count
total_users = user_repo.count()
print(f"Total users: {total_users}")
# Delete a user
user_repo.delete(charlie.id)
print(f"Users after deleting Charlie: {[u.name for u in user_repo.get_collection()]}")
# Check existence
print(f"Does Alice exist? {user_repo.exists(alice.id)}")
print(f"Does Charlie exist? {user_repo.exists(charlie.id)}")
```
### Complete Asynchronous Example
This example demonstrates how to implement and use an asynchronous repository with `AsyncListBasedCrudRepository`.
```python
import asyncio
from abstractrepo.repo import AsyncListBasedCrudRepository
from abstractrepo.specification import AttributeSpecification, Operator
from abstractrepo.order import OrderOptions, OrderOption, OrderDirection
from abstractrepo.paging import PagingOptions
from pydantic import BaseModel
from typing import List, Type, Optional
# Define your data model
class User(BaseModel):
id: int
name: str
email: str
# Define forms for creation and update
class UserCreateForm(BaseModel):
name: str
email: str
class UserUpdateForm(BaseModel):
name: str | None = None
email: str | None = None
# Implement your concrete asynchronous repository
class ConcreteAsyncUserRepository(AsyncListBasedCrudRepository[User, int, UserCreateForm, UserUpdateForm]):
_next_id: int = 0
def __init__(self, items: List[User] | None = None):
super().__init__(items)
if items:
self._next_id = max(item.id for item in items) + 1
@property
def model_class(self) -> Type[User]:
return User
async def _create_model(self, form: UserCreateForm, new_id: int) -> User:
# Simulate async operation
await asyncio.sleep(0.01)
return User(id=new_id, name=form.name, email=form.email)
async def _update_model(self, model: User, form: UserUpdateForm) -> User:
# Simulate async operation
await asyncio.sleep(0.01)
if form.name is not None:
model.name = form.name
if form.email is not None:
model.email = form.email
return model
async def _generate_id(self) -> int:
# Simulate async operation
await asyncio.sleep(0.01)
self._next_id += 1
return self._next_id
def _get_id_filter_specification(self, item_id: int) -> AttributeSpecification[User, bool]:
return AttributeSpecification("id", item_id, Operator.E)
# --- Usage Example ---
async def main():
# Initialize asynchronous repository
async_user_repo = ConcreteAsyncUserRepository()
# Create users asynchronously
alice = await async_user_repo.create(UserCreateForm(name="Alice", email="alice@example.com"))
bob = await async_user_repo.create(UserCreateForm(name="Bob", email="bob@example.com"))
charlie = await async_user_repo.create(UserCreateForm(name="Charlie", email="charlie@example.com"))
print(f"Created users: {alice.name}, {bob.name}, {charlie.name}")
# Query with specifications asynchronously
b_users = await async_user_repo.get_collection(
filter_spec=AttributeSpecification("name", "B%", Operator.LIKE),
order_options=OrderOptions(OrderOption("email", OrderDirection.ASC)),
paging_options=PagingOptions(limit=5)
)
print(f"Users with name starting with 'B': {[u.name for u in b_users]}")
# Get a single user by ID asynchronously
retrieved_alice = await async_user_repo.get_item(alice.id)
print(f"Retrieved user by ID: {retrieved_alice.name}")
# Update a user asynchronously
await async_user_repo.update(bob.id, UserUpdateForm(email="robert@example.com"))
updated_bob = await async_user_repo.get_item(bob.id)
print(f"Updated Bob's email to: {updated_bob.email}")
# Get count asynchronously
total_users = await async_user_repo.count()
print(f"Total users: {total_users}")
# Delete a user asynchronously
await async_user_repo.delete(charlie.id)
users_after_delete = await async_user_repo.get_collection()
print(f"Users after deleting Charlie: {[u.name for u in users_after_delete]}")
# Check existence asynchronously
print(f"Does Alice exist? {await async_user_repo.exists(alice.id)}")
print(f"Does Charlie exist? {await async_user_repo.exists(charlie.id)}")
if __name__ == "__main__":
asyncio.run(main())
```
## API Reference
### Repository Methods
| Method | Parameters | Returns | Description |
|:-----------------|:-------------------------------------------------|:---------------|:-------------------------------------|
| `get_collection` | `filter_spec`, `order_options`, `paging_options` | `List[TModel]` | Get filtered/sorted/paged collection |
| `count` | `filter_spec` | `int` | Count filtered items |
| `get_item` | `item_id` | `TModel` | Get single item by ID |
| `exists` | `item_id` | `bool` | Check item existence |
| `create` | `form` | `TModel` | Create new item |
| `update` | `item_id`, `form` | `TModel` | Update existing item |
| `delete` | `item_id` | `TModel` | Delete item |
### Specification Types
| Class | Description |
|:-------------------------|:--------------------------|
| `AttributeSpecification` | Filter by model attribute |
| `AndSpecification` | Logical AND combination |
| `OrSpecification` | Logical OR combination |
| `NotSpecification` | Logical negation |
### Ordering Options
```
OrderOption(
attribute: str,
direction: OrderDirection = OrderDirection.ASC,
nones: Optional[NonesOrder] = None,
)
```
### Pagination Options
```
PagingOptions(
limit: Optional[int] = None,
offset: Optional[int] = None,
)
```
## Best Practices
1. **Type Safety**: Leverage Python's typing system for robust implementations.
2. **Specification Composition**: Combine simple specs for complex queries.
3. **Null Handling**: Explicitly define null ordering behavior.
4. **Pagination**: Use `PageResolver` for consistent page-based navigation.
5. **Error Handling**: Catch repository-specific exceptions.
6. **Asynchronous Operations**: Use `await` with asynchronous repository methods to ensure non-blocking execution.
7. **Use pydantic for data modeling**: Define your models using pydantic, allowing for robust data validation.
## Dependencies
* Python 3.7+
* No external dependencies
## License
This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for more information.
Raw data
{
"_id": null,
"home_page": "https://github.com/Smoren/abstractrepo-pypi",
"name": "abstractrepo",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": null,
"keywords": "repo, repository, crud",
"author": "Smoren",
"author_email": "ofigate@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/ce/74/5fcc7933d9f5366b9dddf6477c17002cfa413a0db1be1152273abd45984d/abstractrepo-1.4.3.tar.gz",
"platform": null,
"description": "# AbstractRepo - Python Repository Pattern Implementation\n\n[](https://pypi.org/project/abstractrepo/)\n[](https://github.com/Smoren/abstractrepo-pypi/releases)\n[](https://coveralls.io/github/Smoren/abstractrepo-pypi?branch=master)\n[](https://pepy.tech/projects/abstractrepo)\n[](https://github.com/Smoren/abstractrepo-pypi/actions)\n[](https://github.com/Smoren/abstractrepo-pypi/blob/master/LICENSE)\n\nThe `AbstractRepo` library provides a robust and flexible abstraction layer for interacting with various data storage systems in Python. It implements the widely recognized Repository Pattern, offering a clean and consistent API for common data operations such as Create, Read, Update, and Delete (CRUD). This design promotes a clear separation of concerns, making your application logic independent of the underlying data persistence mechanism. This allows for easy switching between different databases or storage solutions without significant changes to your core business logic.\n\n## Key Features\n\n* **CRUD Operations:** Comprehensive support for standard data manipulation operations.\n* **Specifications Pattern:** A powerful and flexible mechanism for defining complex query criteria based on business rules, enabling highly customizable data retrieval.\n* **Ordering Options:** Advanced sorting capabilities, including control over the placement of `NULL` values.\n* **Pagination Support:** Efficient handling of large datasets through limit and offset-based pagination.\n* **Strong Typing:** Leverages Python's type hinting for improved code readability, maintainability, and early error detection.\n* **Extensibility:** Designed with extensibility in mind, allowing for easy integration with various database technologies and custom data sources.\n* **In-Memory Implementation:** Includes a built-in list-based repository implementation, ideal for testing, development, and rapid prototyping.\n* **Asynchronous Support:** Provides interfaces and base implementations for asynchronous repository operations, crucial for modern, high-performance applications.\n\n## Implementations\n\n* [SQLAlchemy implementation](https://github.com/Smoren/abstractrepo-sqlalchemy-pypi)\n* [List-based in-memory implementation](#list-based-implementation-listbasedcrudrepository)\n\n## Installation\n\nTo get started with `AbstractRepo`, install it using pip:\n\n```shell\npip install abstractrepo\n```\n\n## Table of Contents\n\n* [Core Components and Usage](#core-components-and-usage)\n * [Repository Interface (`CrudRepositoryInterface`)](#repository-interface-crudrepositoryinterface)\n * [List-Based Implementation (`ListBasedCrudRepository`)](#list-based-implementation-listbasedcrudrepository)\n * [Asynchronous Repositories (`AsyncCrudRepositoryInterface`)](#asynchronous-repositories)\n * [Specifications](#specifications)\n * [Ordering](#ordering)\n * [Pagination](#pagination)\n * [Exception Handling](#exception-handling)\n* [Examples](#examples)\n * [Complete Synchronous Example](#complete-synchronous-example)\n * [Complete Asynchronous Example](#complete-asynchronous-example)\n* [API Reference](#api-reference)\n * [Repository Methods](#repository-methods)\n * [Specification Types](#specification-types)\n * [Ordering Options](#ordering-options)\n * [Pagination Options](#pagination-options)\n* [Best Practices](#best-practices)\n* [Dependencies](#dependencies)\n* [License](#license)\n\n## Core Components and Usage\n\n### Repository Interface (`CrudRepositoryInterface`)\n\nThe `CrudRepositoryInterface` defines the contract for all synchronous repositories. It specifies the standard CRUD operations and other essential methods that any concrete repository implementation must adhere to.\n\n```python\nimport abc\nfrom pydantic import BaseModel\nfrom abstractrepo.repo import CrudRepositoryInterface\n\n\nclass User(BaseModel):\n id: int\n username: str\n password: str\n display_name: str\n\nclass UserCreateForm(BaseModel):\n username: str\n password: str\n display_name: str\n\nclass UserUpdateForm(BaseModel):\n display_name: str\n\n\nclass UserRepositoryInterface(CrudRepositoryInterface[User, int, UserCreateForm, UserUpdateForm], abc.ABC):\n pass\n\n\nclass UserRepository(UserRepositoryInterface):\n # Implement abstract methods here\n ...\n```\n\n**Key Methods:**\n\n| Method | Parameters | Returns | Description |\n|:-----------------|:-------------------------------------------------|:---------------|:-----------------------------------------------------------------------------------------------|\n| `get_collection` | `filter_spec`, `order_options`, `paging_options` | `List[TModel]` | Retrieves a collection of items based on filtering, sorting, and pagination options. |\n| `count` | `filter_spec` | `int` | Returns the total count of items matching the given filter specification. |\n| `get_item` | `item_id` | `TModel` | Retrieves a single item by its unique identifier. Raises `ItemNotFoundException` if not found. |\n| `exists` | `item_id` | `bool` | Checks if an item with the specified ID exists in the repository. |\n| `create` | `form` | `TModel` | Creates a new item in the repository using the provided creation form. |\n| `update` | `item_id`, `form` | `TModel` | Updates an existing item identified by its ID with data from the update form. |\n| `delete` | `item_id` | `TModel` | Deletes an item from the repository by its ID. |\n| `model_class` | (Property) | `Type[TModel]` | Returns the model class associated with the repository. |\n\n### List-Based Implementation (`ListBasedCrudRepository`)\n\nThe `ListBasedCrudRepository` provides a concrete, in-memory implementation of the `CrudRepositoryInterface`. It's particularly useful for testing, development, and scenarios where a simple, non-persistent data store is sufficient.\n\n```python\nimport abc\nfrom typing import Optional, List, Type\nfrom pydantic import BaseModel\nfrom abstractrepo.repo import CrudRepositoryInterface, ListBasedCrudRepository\nfrom abstractrepo.specification import SpecificationInterface, AttributeSpecification, Operator\nfrom abstractrepo.exceptions import ItemNotFoundException, UniqueViolationException\n\n\nclass User(BaseModel):\n id: int\n username: str\n password: str\n display_name: str\n\nclass UserCreateForm(BaseModel):\n username: str\n password: str\n display_name: str\n\nclass UserUpdateForm(BaseModel):\n display_name: str\n\n\nclass UserRepositoryInterface(CrudRepositoryInterface[User, int, UserCreateForm, UserUpdateForm], abc.ABC):\n pass\n\n\nclass ListBasedUserRepository(\n ListBasedCrudRepository[User, int, UserCreateForm, UserUpdateForm],\n UserRepositoryInterface,\n):\n _next_id: int\n\n def __init__(self, items: Optional[List[User]] = None):\n super().__init__(items)\n self._next_id = 0\n\n def get_by_username(self, username: str) -> User:\n items = self.get_collection(AttributeSpecification(\"username\", username))\n if len(items) == 0:\n raise ItemNotFoundException(User)\n\n return items[0]\n\n @property\n def model_class(self) -> Type[User]:\n return User\n\n def _create_model(self, form: UserCreateForm, new_id: int) -> User:\n if self._username_exists(form.username):\n raise UniqueViolationException(User, \"create\", form)\n\n return User(\n id=new_id,\n username=form.username,\n password=form.password,\n display_name=form.display_name,\n )\n\n def _update_model(self, model: User, form: UserUpdateForm) -> User:\n model.display_name = form.display_name\n return model\n\n def _username_exists(self, username: str) -> bool:\n return self.count(AttributeSpecification(\"username\", username)) > 0\n\n def _generate_id(self) -> int:\n self._next_id += 1\n return self._next_id\n\n def _get_id_filter_specification(self, item_id: int) -> SpecificationInterface[User, bool]:\n return AttributeSpecification(\"id\", item_id, Operator.E)\n```\n\n### Asynchronous Repositories\n\n`AbstractRepo` provides full support for asynchronous operations, allowing you to build non-blocking data access layers for high-performance applications. The `AsyncCrudRepositoryInterface` defines the asynchronous contract, and `AsyncListBasedCrudRepository` offers an in-memory asynchronous implementation.\n\nThis interface mirrors the synchronous `CrudRepositoryInterface` but with `async` methods, enabling seamless integration with `asyncio` and other asynchronous frameworks.\n\n```python\nimport abc\nfrom typing import TypeVar\nfrom pydantic import BaseModel\nfrom abstractrepo.repo import AsyncCrudRepositoryInterface\n\n\nclass User(BaseModel):\n id: int\n username: str\n password: str\n display_name: str\n\nclass UserCreateForm(BaseModel):\n username: str\n password: str\n display_name: str\n\nclass UserUpdateForm(BaseModel):\n display_name: str\n\n\nclass AsyncUserRepositoryInterface(AsyncCrudRepositoryInterface[User, int, UserCreateForm, UserUpdateForm], abc.ABC):\n pass\n\n\nclass AsyncUserRepository(AsyncUserRepositoryInterface):\n # Implement abstract async methods here\n ...\n```\n\n### Specifications\n\nThe Specifications Pattern allows you to define flexible and reusable filtering logic. You can combine simple specifications to build complex queries.\n\n**Filtering with Specifications:**\n\n```python\nfrom abstractrepo.specification import AttributeSpecification, AndSpecification, OrSpecification, Operator\n\n# Single attribute filter\nactive_users_spec = AttributeSpecification(\"is_active\", True)\n\n# Complex filter combining AND and OR operations\npremium_filter_spec = AndSpecification(\n AttributeSpecification(\"plan\", \"premium\"),\n OrSpecification(\n AttributeSpecification(\"age\", 30, Operator.GTE),\n AttributeSpecification(\"join_date\", \"2023-01-01\", Operator.GT)\n )\n)\n```\n\n**Supported Operators:**\n\n`AbstractRepo` provides a rich set of operators for various comparison and matching needs:\n\n| Operator | Description |\n| :------- |:-------------------------------|\n| `E` | Equal |\n| `NE` | Not Equal |\n| `GT` | Greater Than |\n| `LT` | Less Than |\n| `GTE` | Greater Than or Equal |\n| `LTE` | Less Than or Equal |\n| `LIKE` | Case-Sensitive Pattern Match |\n| `ILIKE` | Case-Insensitive Pattern Match |\n| `IN` | In List |\n| `NOT_IN` | Not In List |\n\n### Ordering\n\nControl the order of retrieved items using `OrderOptions` and `OrderOption`. You can specify the attribute to sort by, the direction (ascending or descending), and how `None` values should be handled.\n\n```python\nfrom abstractrepo.order import OrderOptionsBuilder, OrderOptions, OrderOption, OrderDirection, NonesOrder\n\n# Single field ordering\nordering_by_name = OrderOptions(\n OrderOption(\"name\", OrderDirection.ASC, NonesOrder.LAST)\n)\n\n# Multi-field ordering using OrderOptionsBuilder for chaining\ncomplex_ordering = OrderOptionsBuilder() \\\n .add(\"priority\", OrderDirection.DESC) \\\n .add(\"created_at\", OrderDirection.ASC, NonesOrder.LAST) \\\n .build()\n```\n\n### Pagination\n\nManage large result sets efficiently with `PagingOptions` for limit/offset-based pagination and `PageResolver` for page-number-based navigation.\n\n```python\nfrom abstractrepo.paging import PagingOptions, PageResolver\n\n# Manual paging with limit and offset\nmanual_paging = PagingOptions(limit=10, offset=20)\n\n# Page-based resolver for consistent page navigation\nresolver = PageResolver(page_size=25)\npage_3_options = resolver.get_page(3) # Retrieves PagingOptions for the 3rd page\n```\n\n### Exception Handling\n\n`AbstractRepo` defines specific exceptions to handle common repository-related errors, allowing for robust error management in your application.\n\n```python\nfrom abstractrepo.exceptions import (\n ItemNotFoundException,\n UniqueViolationException,\n RelationViolationException,\n)\n\ntry:\n repo.get_item(999) # Attempt to retrieve a non-existent item\nexcept ItemNotFoundException as e:\n print(f\"Error: {e}\") # Handle the case where the item is not found\n```\n\n## Examples\n\n### Complete Synchronous Example\n\nThis example demonstrates the full lifecycle of a synchronous repository using the `ListBasedCrudRepository`.\n\n```python\nfrom abstractrepo.repo import ListBasedCrudRepository\nfrom abstractrepo.specification import AttributeSpecification, Operator\nfrom abstractrepo.order import OrderOptions, OrderOption, OrderDirection\nfrom abstractrepo.paging import PagingOptions\nfrom pydantic import BaseModel\n\n# Define your data model\nclass User(BaseModel):\n id: int\n name: str\n email: str\n\n# Define forms for creation and update (can be pure classes Pydantic models)\nclass UserCreateForm(BaseModel):\n name: str\n email: str\n\nclass UserUpdateForm(BaseModel):\n name: str | None = None\n email: str | None = None\n\n# Implement your concrete repository\nclass ConcreteUserRepository(ListBasedCrudRepository[User, int, UserCreateForm, UserUpdateForm]):\n _next_id: int = 0\n\n def __init__(self, items: list[User] | None = None):\n super().__init__(items)\n if items:\n self._next_id = max(item.id for item in items) + 1\n\n @property\n def model_class(self) -> type[User]:\n return User\n\n def _create_model(self, form: UserCreateForm, new_id: int) -> User:\n # In a real scenario, you might check for unique constraints here\n return User(id=new_id, name=form.name, email=form.email)\n\n def _update_model(self, model: User, form: UserUpdateForm) -> User:\n if form.name is not None:\n model.name = form.name\n if form.email is not None:\n model.email = form.email\n return model\n\n def _generate_id(self) -> int:\n self._next_id += 1\n return self._next_id\n\n def _get_id_filter_specification(self, item_id: int) -> AttributeSpecification[User, bool]:\n return AttributeSpecification(\"id\", item_id, Operator.E)\n\n\n# --- Usage Example ---\n\n# Initialize repository\nuser_repo = ConcreteUserRepository()\n\n# Create users\nalice = user_repo.create(UserCreateForm(name=\"Alice\", email=\"alice@example.com\"))\nbob = user_repo.create(UserCreateForm(name=\"Bob\", email=\"bob@example.com\"))\ncharlie = user_repo.create(UserCreateForm(name=\"Charlie\", email=\"charlie@example.com\"))\n\nprint(f\"Created users: {alice.name}, {bob.name}, {charlie.name}\")\n\n# Query with specifications\nb_users = user_repo.get_collection(\n filter_spec=AttributeSpecification(\"name\", \"B%\", Operator.LIKE),\n order_options=OrderOptions(OrderOption(\"email\", OrderDirection.ASC)),\n paging_options=PagingOptions(limit=5)\n)\nprint(f\"Users with name starting with 'B': {[u.name for u in b_users]}\")\n\n# Get a single user by ID\nretrieved_alice = user_repo.get_item(alice.id)\nprint(f\"Retrieved user by ID: {retrieved_alice.name}\")\n\n# Update a user\nuser_repo.update(bob.id, UserUpdateForm(email=\"robert@example.com\"))\nupdated_bob = user_repo.get_item(bob.id)\nprint(f\"Updated Bob's email to: {updated_bob.email}\")\n\n# Get count\ntotal_users = user_repo.count()\nprint(f\"Total users: {total_users}\")\n\n# Delete a user\nuser_repo.delete(charlie.id)\nprint(f\"Users after deleting Charlie: {[u.name for u in user_repo.get_collection()]}\")\n\n# Check existence\nprint(f\"Does Alice exist? {user_repo.exists(alice.id)}\")\nprint(f\"Does Charlie exist? {user_repo.exists(charlie.id)}\")\n```\n\n### Complete Asynchronous Example\n\nThis example demonstrates how to implement and use an asynchronous repository with `AsyncListBasedCrudRepository`.\n\n```python\nimport asyncio\nfrom abstractrepo.repo import AsyncListBasedCrudRepository\nfrom abstractrepo.specification import AttributeSpecification, Operator\nfrom abstractrepo.order import OrderOptions, OrderOption, OrderDirection\nfrom abstractrepo.paging import PagingOptions\nfrom pydantic import BaseModel\nfrom typing import List, Type, Optional\n\n# Define your data model\nclass User(BaseModel):\n id: int\n name: str\n email: str\n\n# Define forms for creation and update\nclass UserCreateForm(BaseModel):\n name: str\n email: str\n\nclass UserUpdateForm(BaseModel):\n name: str | None = None\n email: str | None = None\n\n# Implement your concrete asynchronous repository\nclass ConcreteAsyncUserRepository(AsyncListBasedCrudRepository[User, int, UserCreateForm, UserUpdateForm]):\n _next_id: int = 0\n\n def __init__(self, items: List[User] | None = None):\n super().__init__(items)\n if items:\n self._next_id = max(item.id for item in items) + 1\n\n @property\n def model_class(self) -> Type[User]:\n return User\n\n async def _create_model(self, form: UserCreateForm, new_id: int) -> User:\n # Simulate async operation\n await asyncio.sleep(0.01)\n return User(id=new_id, name=form.name, email=form.email)\n\n async def _update_model(self, model: User, form: UserUpdateForm) -> User:\n # Simulate async operation\n await asyncio.sleep(0.01)\n if form.name is not None:\n model.name = form.name\n if form.email is not None:\n model.email = form.email\n return model\n\n async def _generate_id(self) -> int:\n # Simulate async operation\n await asyncio.sleep(0.01)\n self._next_id += 1\n return self._next_id\n\n def _get_id_filter_specification(self, item_id: int) -> AttributeSpecification[User, bool]:\n return AttributeSpecification(\"id\", item_id, Operator.E)\n\n\n# --- Usage Example ---\n\nasync def main():\n # Initialize asynchronous repository\n async_user_repo = ConcreteAsyncUserRepository()\n\n # Create users asynchronously\n alice = await async_user_repo.create(UserCreateForm(name=\"Alice\", email=\"alice@example.com\"))\n bob = await async_user_repo.create(UserCreateForm(name=\"Bob\", email=\"bob@example.com\"))\n charlie = await async_user_repo.create(UserCreateForm(name=\"Charlie\", email=\"charlie@example.com\"))\n\n print(f\"Created users: {alice.name}, {bob.name}, {charlie.name}\")\n\n # Query with specifications asynchronously\n b_users = await async_user_repo.get_collection(\n filter_spec=AttributeSpecification(\"name\", \"B%\", Operator.LIKE),\n order_options=OrderOptions(OrderOption(\"email\", OrderDirection.ASC)),\n paging_options=PagingOptions(limit=5)\n )\n print(f\"Users with name starting with 'B': {[u.name for u in b_users]}\")\n\n # Get a single user by ID asynchronously\n retrieved_alice = await async_user_repo.get_item(alice.id)\n print(f\"Retrieved user by ID: {retrieved_alice.name}\")\n\n # Update a user asynchronously\n await async_user_repo.update(bob.id, UserUpdateForm(email=\"robert@example.com\"))\n updated_bob = await async_user_repo.get_item(bob.id)\n print(f\"Updated Bob's email to: {updated_bob.email}\")\n\n # Get count asynchronously\n total_users = await async_user_repo.count()\n print(f\"Total users: {total_users}\")\n\n # Delete a user asynchronously\n await async_user_repo.delete(charlie.id)\n users_after_delete = await async_user_repo.get_collection()\n print(f\"Users after deleting Charlie: {[u.name for u in users_after_delete]}\")\n\n # Check existence asynchronously\n print(f\"Does Alice exist? {await async_user_repo.exists(alice.id)}\")\n print(f\"Does Charlie exist? {await async_user_repo.exists(charlie.id)}\")\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n## API Reference\n\n### Repository Methods\n\n| Method | Parameters | Returns | Description |\n|:-----------------|:-------------------------------------------------|:---------------|:-------------------------------------|\n| `get_collection` | `filter_spec`, `order_options`, `paging_options` | `List[TModel]` | Get filtered/sorted/paged collection |\n| `count` | `filter_spec` | `int` | Count filtered items |\n| `get_item` | `item_id` | `TModel` | Get single item by ID |\n| `exists` | `item_id` | `bool` | Check item existence |\n| `create` | `form` | `TModel` | Create new item |\n| `update` | `item_id`, `form` | `TModel` | Update existing item |\n| `delete` | `item_id` | `TModel` | Delete item |\n\n### Specification Types\n\n| Class | Description |\n|:-------------------------|:--------------------------|\n| `AttributeSpecification` | Filter by model attribute |\n| `AndSpecification` | Logical AND combination |\n| `OrSpecification` | Logical OR combination |\n| `NotSpecification` | Logical negation |\n\n### Ordering Options\n\n```\nOrderOption(\n attribute: str,\n direction: OrderDirection = OrderDirection.ASC,\n nones: Optional[NonesOrder] = None,\n)\n```\n\n### Pagination Options\n\n```\nPagingOptions(\n limit: Optional[int] = None,\n offset: Optional[int] = None,\n)\n```\n\n## Best Practices\n\n1. **Type Safety**: Leverage Python's typing system for robust implementations.\n2. **Specification Composition**: Combine simple specs for complex queries.\n3. **Null Handling**: Explicitly define null ordering behavior.\n4. **Pagination**: Use `PageResolver` for consistent page-based navigation.\n5. **Error Handling**: Catch repository-specific exceptions.\n6. **Asynchronous Operations**: Use `await` with asynchronous repository methods to ensure non-blocking execution.\n7. **Use pydantic for data modeling**: Define your models using pydantic, allowing for robust data validation.\n\n## Dependencies\n\n* Python 3.7+\n* No external dependencies\n\n## License\n\nThis project is licensed under the MIT License. See the [LICENSE](LICENSE) file for more information.\n",
"bugtrack_url": null,
"license": null,
"summary": "Abstract CRUD repository components",
"version": "1.4.3",
"project_urls": {
"Bug Reports": "https://github.com/Smoren/abstractrepo-pypi/issues",
"Documentation": "https://github.com/Smoren/abstractrepo-pypi",
"Homepage": "https://github.com/Smoren/abstractrepo-pypi",
"Source Code": "https://github.com/Smoren/abstractrepo-pypi"
},
"split_keywords": [
"repo",
" repository",
" crud"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "f69c4b91b5a109b54313ca9b95678974e847b3831ca1501d90294f8c45e3046a",
"md5": "26404fdf7ed9bb8855817af0d36c9c00",
"sha256": "46680c07d31b034c4724f795aadec58a635d34719a642a73f965d0d6b86da235"
},
"downloads": -1,
"filename": "abstractrepo-1.4.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "26404fdf7ed9bb8855817af0d36c9c00",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 18357,
"upload_time": "2025-08-24T10:33:59",
"upload_time_iso_8601": "2025-08-24T10:33:59.029814Z",
"url": "https://files.pythonhosted.org/packages/f6/9c/4b91b5a109b54313ca9b95678974e847b3831ca1501d90294f8c45e3046a/abstractrepo-1.4.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "ce745fcc7933d9f5366b9dddf6477c17002cfa413a0db1be1152273abd45984d",
"md5": "200757636678afa07d7eaa0eb56989fd",
"sha256": "f2da9454b63dd35d6a2b8b6c4fbe545bd1d99e0e10c37ad011278b182c0af709"
},
"downloads": -1,
"filename": "abstractrepo-1.4.3.tar.gz",
"has_sig": false,
"md5_digest": "200757636678afa07d7eaa0eb56989fd",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 22815,
"upload_time": "2025-08-24T10:34:00",
"upload_time_iso_8601": "2025-08-24T10:34:00.093391Z",
"url": "https://files.pythonhosted.org/packages/ce/74/5fcc7933d9f5366b9dddf6477c17002cfa413a0db1be1152273abd45984d/abstractrepo-1.4.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-24 10:34:00",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Smoren",
"github_project": "abstractrepo-pypi",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"tox": true,
"lcname": "abstractrepo"
}
JSON
