Overview
AbstractRepository is an abstract base class that provides standard CRUD operations for entities. It offers methods for finding, filtering, and querying entities with a clean, consistent interface.
Class Definition
from framefox.core.orm import AbstractRepository
from typing import Type, TypeVar
from sqlmodel import SQLModel
T = TypeVar("T", bound=SQLModel)
class AbstractRepository(ABC):
def __init__(self, model: Type[T]):
self.model = model
self.create_model = self.model.generate_create_model()
Constructor
init
Initializes a repository for a specific entity model.
def __init__(self, model: Type[T])
The entity class this repository will manage
Example
from framefox.core.orm import AbstractRepository, AbstractEntity
from sqlmodel import Field
class User(AbstractEntity, table=True):
id: int | None = Field(default=None, primary_key=True)
username: str
email: str
class UserRepository(AbstractRepository):
def __init__(self):
super().__init__(User)
user_repo = UserRepository()
Properties
entity_manager
Returns the entity manager for the current request context.
@property
def entity_manager(self) -> EntityManager
The entity manager instance for the current request
Methods
find
Retrieves an entity by its primary key ID.
def find(self, id) -> Optional[T]
The primary key value of the entity to retrieve
The retrieved entity, or None if not found
Example
user_repo = UserRepository()
user = user_repo.find(123)
if user:
print(f"Found user: {user.username}")
else:
print("User not found")
find_all
Retrieves all entities of the repository’s type.
def find_all(self) -> List[T]
Example
user_repo = UserRepository()
all_users = user_repo.find_all()
for user in all_users:
print(user.username)
find_by
Retrieves entities based on specific criteria with optional ordering, limiting, and pagination.
def find_by(
self,
criteria,
order_by=None,
limit=None,
offset=None
) -> List[T]
A dictionary of field names and values to filter by
A dictionary mapping field names to sort direction ("asc" or "desc")
Maximum number of entities to retrieve
Number of entities to skip (for pagination)
A list of entities matching the criteria
Example
user_repo = UserRepository()
# Find users with specific criteria
active_users = user_repo.find_by(
criteria={"status": "active"},
order_by={"created_at": "desc"},
limit=10,
offset=0
)
# Find users by multiple criteria
admin_users = user_repo.find_by(
criteria={"role": "admin", "is_active": True}
)
find_one_by
Retrieves a single entity based on specific criteria.
def find_one_by(self, criteria) -> Optional[T]
A dictionary of field names and values to filter by
The first entity that matches the criteria, or None if not found
Example
user_repo = UserRepository()
# Find a single user by username
user = user_repo.find_one_by(criteria={"username": "john_doe"})
if user:
print(f"User email: {user.email}")
# Find by multiple criteria
admin_user = user_repo.find_one_by(
criteria={"email": "[email protected]", "role": "admin"}
)
get_query_builder
Returns a QueryBuilder instance configured for the repository’s entity, allowing for complex query construction.
def get_query_builder(self) -> QueryBuilder
A QueryBuilder instance configured for this repository’s entity
Example
user_repo = UserRepository()
# Get query builder for complex queries
query_builder = user_repo.get_query_builder()
# Build and execute a complex query
results = query_builder.select() \
.where(User.age >= 18) \
.where(User.status == "active") \
.order_by(User.created_at.desc()) \
.limit(20) \
.execute()
Complete Example
from framefox.core.orm import AbstractRepository, AbstractEntity
from sqlmodel import Field
class Product(AbstractEntity, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
category: str
price: float
stock: int
is_active: bool = True
class ProductRepository(AbstractRepository):
def __init__(self):
super().__init__(Product)
def find_by_category(self, category: str):
"""Custom method: Find products by category"""
return self.find_by(criteria={"category": category})
def find_in_stock(self):
"""Custom method: Find products with stock > 0"""
qb = self.get_query_builder()
return qb.select() \
.where(Product.stock > 0) \
.where(Product.is_active == True) \
.execute()
# Usage
product_repo = ProductRepository()
# Find by ID
product = product_repo.find(1)
# Find all products
all_products = product_repo.find_all()
# Find with criteria and sorting
expensive_products = product_repo.find_by(
criteria={"is_active": True},
order_by={"price": "desc"},
limit=10
)
# Find one by criteria
laptop = product_repo.find_one_by(criteria={"name": "Laptop"})
# Use custom methods
electronics = product_repo.find_by_category("Electronics")
in_stock_products = product_repo.find_in_stock()
# Complex query with query builder
qb = product_repo.get_query_builder()
affordable_electronics = qb.select() \
.where(Product.category == "Electronics") \
.where(Product.price < 1000) \
.where(Product.stock > 5) \
.order_by(Product.price.asc()) \
.limit(20) \
.execute()
Notes
- Repositories provide a clean abstraction layer over database operations
- Extend
AbstractRepository to create custom repositories with domain-specific methods
- The
entity_manager property automatically retrieves the request-scoped entity manager
- Use
get_query_builder() for complex queries that go beyond simple filtering
- All query methods return lists except
find() and find_one_by() which return single entities or None