Repositories

Repositories in Framefox provide a clean abstraction layer for database operations. They encapsulate common CRUD operations and allow you to add custom query methods.

AbstractRepository Class

The AbstractRepository class provides the foundation for all repositories:

from abc import ABC
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()

Reference: framefox/core/orm/abstract_repository.py:20

Entity Manager Access

Repositories automatically access the request-scoped EntityManager:

@property
def entity_manager(self):
    return EntityManagerRegistry.get_entity_manager_for_request()

Reference: framefox/core/orm/abstract_repository.py:38

Creating a Repository

Create a repository by extending AbstractRepository:

from framefox.core.orm.abstract_repository import AbstractRepository
from myapp.models import User

class UserRepository(AbstractRepository):
    def __init__(self):
        super().__init__(User)

Register with Dependency Injection

from framefox.core.di.service_container import ServiceContainer

container = ServiceContainer()
container.register(
    UserRepository,
    tags=["repository.user"]
)

CRUD Operations

`find(id)` - Find by ID

Retrieve an entity by its primary key:

def find(self, id) -> Optional[T]:
    return self.entity_manager.find(self.model, id)

Reference: framefox/core/orm/abstract_repository.py:42 Usage:

user_repo = UserRepository()

# Find by single primary key
user = user_repo.find(1)

if user:
    print(f"Found user: {user.name}")
else:
    print("User not found")

`find_all()` - Get All Records

Retrieve all entities of a type:

def find_all(self) -> List[T]:
    statement = select(self.model)
    return self.entity_manager.exec_statement(statement)

Reference: framefox/core/orm/abstract_repository.py:54 Usage:

user_repo = UserRepository()
all_users = user_repo.find_all()

for user in all_users:
    print(f"{user.id}: {user.name}")

`find_by()` - Find with Criteria

Retrieve 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]:
    statement = select(self.model).filter_by(**criteria)
    
    if order_by:
        for field, direction in order_by.items():
            if direction.lower() == "asc":
                statement = statement.order_by(asc(getattr(self.model, field)))
            elif direction.lower() == "desc":
                statement = statement.order_by(desc(getattr(self.model, field)))
    
    if limit is not None:
        statement = statement.limit(limit)
    
    if offset is not None:
        statement = statement.offset(offset)
    
    return self.entity_manager.exec_statement(statement)

Reference: framefox/core/orm/abstract_repository.py:64 Usage:

user_repo = UserRepository()

# Simple filter
active_users = user_repo.find_by({"is_active": True})

# With ordering
users = user_repo.find_by(
    {"is_active": True},
    order_by={"created_at": "desc"}
)

# With pagination
users = user_repo.find_by(
    {"is_active": True},
    order_by={"name": "asc"},
    limit=10,
    offset=20  # Page 3 (if 10 per page)
)

# Multiple criteria
admin_users = user_repo.find_by({
    "role": "admin",
    "is_active": True
})

`find_one_by()` - Find Single Record

Retrieve the first entity matching criteria:

def find_one_by(self, criteria) -> Optional[T]:
    statement = select(self.model).filter_by(**criteria).limit(1)
    results = self.entity_manager.exec_statement(statement)
    return results[0] if results else None

Reference: framefox/core/orm/abstract_repository.py:92 Usage:

user_repo = UserRepository()

# Find by email
user = user_repo.find_one_by({"email": "[email protected]"})

if user:
    print(f"Found: {user.name}")

Save Operations

To save (create or update) entities, use the EntityManager directly:

user_repo = UserRepository()

# Create new user
user = User(name="Alice", email="[email protected]")
user_repo.entity_manager.persist(user)
user_repo.entity_manager.commit()

# Update existing user
user = user_repo.find(1)
user.name = "Alice Updated"
user_repo.entity_manager.persist(user)
user_repo.entity_manager.commit()

Delete Operations

Delete entities using the EntityManager:

user_repo = UserRepository()

# Delete by ID
user = user_repo.find(1)
if user:
    user_repo.entity_manager.delete(user)
    user_repo.entity_manager.commit()

Custom Repository Methods

Extend repositories with custom methods for your specific needs:

class UserRepository(AbstractRepository):
    def __init__(self):
        super().__init__(User)
    
    def find_active_users(self) -> List[User]:
        """Get all active users"""
        return self.find_by({"is_active": True})
    
    def find_by_email(self, email: str) -> Optional[User]:
        """Find user by email address"""
        return self.find_one_by({"email": email})
    
    def find_admins(self) -> List[User]:
        """Get all admin users"""
        return self.find_by(
            {"role": "admin", "is_active": True},
            order_by={"name": "asc"}
        )
    
    def count_active_users(self) -> int:
        """Count active users"""
        from sqlmodel import select, func
        statement = select(func.count(User.id)).where(User.is_active == True)
        result = self.entity_manager.exec_statement(statement)
        return result[0] if result else 0
    
    def find_users_registered_after(self, date) -> List[User]:
        """Find users registered after a specific date"""
        from sqlmodel import select
        statement = select(User).where(User.created_at > date)
        return self.entity_manager.exec_statement(statement)

Advanced Custom Methods

For complex queries, use the QueryBuilder:

class OrderRepository(AbstractRepository):
    def __init__(self):
        super().__init__(Order)
    
    def find_high_value_orders(self, min_amount: float) -> List[Order]:
        """Find orders above a certain amount"""
        qb = self.get_query_builder()
        return qb.select() \
            .where(Order.total_amount >= min_amount) \
            .order_by(Order.created_at.desc()) \
            .execute()
    
    def find_orders_with_customer(self, customer_id: int) -> List[Order]:
        """Find all orders for a customer with items"""
        qb = self.get_query_builder()
        return qb.select() \
            .where(Order.customer_id == customer_id) \
            .order_by(Order.created_at.desc()) \
            .execute()

Query Builder Integration

Repositories provide access to the QueryBuilder for complex queries:

def get_query_builder(self) -> QueryBuilder:
    return QueryBuilder(entity_manager=self.entity_manager, model=self.model)

Reference: framefox/core/orm/abstract_repository.py:106 Usage:

user_repo = UserRepository()
qb = user_repo.get_query_builder()

# Build complex query
users = qb.select() \
    .where(User.age >= 18) \
    .where(User.is_active == True) \
    .order_by(User.name.asc()) \
    .limit(10) \
    .execute()

Transaction Support

Use transactions for multiple operations:

user_repo = UserRepository()

with user_repo.entity_manager.transaction():
    # Create user
    user = User(name="Bob", email="[email protected]")
    user_repo.entity_manager.persist(user)
    
    # Create related profile
    profile = Profile(user_id=user.id, bio="Software developer")
    user_repo.entity_manager.persist(profile)
    
    # Both operations commit together or roll back on error

Best Practices

One Repository per Entity: Create a dedicated repository for each main entity
Encapsulate Business Logic: Put domain-specific queries in custom methods
Use Descriptive Method Names: Name methods clearly (e.g., find_active_users not get_users)
Return Type Hints: Always specify return types for better IDE support
Handle None Results: Check for None when using find() or find_one_by()
Use Transactions: Wrap multiple operations in transactions for consistency
Leverage QueryBuilder: Use QueryBuilder for complex queries instead of raw SQL

Complete Example

from typing import List, Optional
from datetime import datetime, timedelta
from framefox.core.orm.abstract_repository import AbstractRepository
from myapp.models import User

class UserRepository(AbstractRepository):
    def __init__(self):
        super().__init__(User)
    
    def create_user(self, name: str, email: str) -> User:
        """Create and persist a new user"""
        user = User(name=name, email=email)
        self.entity_manager.persist(user)
        self.entity_manager.commit()
        return user
    
    def find_by_email(self, email: str) -> Optional[User]:
        """Find user by email"""
        return self.find_one_by({"email": email})
    
    def find_recent_users(self, days: int = 7) -> List[User]:
        """Find users created in the last N days"""
        cutoff = datetime.now() - timedelta(days=days)
        qb = self.get_query_builder()
        return qb.select() \
            .where(User.created_at >= cutoff) \
            .order_by(User.created_at.desc()) \
            .execute()
    
    def update_user(self, user_id: int, **kwargs) -> Optional[User]:
        """Update user fields"""
        user = self.find(user_id)
        if not user:
            return None
        
        for key, value in kwargs.items():
            if hasattr(user, key):
                setattr(user, key, value)
        
        self.entity_manager.persist(user)
        self.entity_manager.commit()
        return user
    
    def delete_user(self, user_id: int) -> bool:
        """Delete a user by ID"""
        user = self.find(user_id)
        if not user:
            return False
        
        self.entity_manager.delete(user)
        self.entity_manager.commit()
        return True

# Usage
user_repo = UserRepository()

# Create
user = user_repo.create_user("Alice", "[email protected]")

# Read
user = user_repo.find_by_email("[email protected]")
recent = user_repo.find_recent_users(days=30)

# Update
user_repo.update_user(user.id, name="Alice Updated")

# Delete
user_repo.delete_user(user.id)

Getting Started

Core Concepts

Controllers

Templates

Database & ORM

Security

Advanced Features

Deployment

AbstractRepository Class

Entity Manager Access

Creating a Repository

Register with Dependency Injection

CRUD Operations

`find(id)` - Find by ID

`find_all()` - Get All Records

`find_by()` - Find with Criteria

`find_one_by()` - Find Single Record

Save Operations

Delete Operations

Custom Repository Methods

Advanced Custom Methods

Query Builder Integration

Transaction Support

Best Practices

Complete Example

Next Steps

Query Builder

Entities

Build docs developers (and LLMs) love

Getting Started

Core Concepts

Controllers

Templates

Database & ORM

Security

Advanced Features

Deployment

​AbstractRepository Class

​Entity Manager Access

​Creating a Repository

​Register with Dependency Injection

​CRUD Operations

​find(id) - Find by ID

​find_all() - Get All Records

​find_by() - Find with Criteria

​find_one_by() - Find Single Record

​Save Operations

​Delete Operations

​Custom Repository Methods

​Advanced Custom Methods

​Query Builder Integration

​Transaction Support

​Best Practices

​Complete Example

​Next Steps

Query Builder

Entities

Build docs developers (and LLMs) love

AbstractRepository Class

Entity Manager Access

Creating a Repository

Register with Dependency Injection

CRUD Operations

`find(id)` - Find by ID

`find_all()` - Get All Records

`find_by()` - Find with Criteria

`find_one_by()` - Find Single Record

Save Operations

Delete Operations

Custom Repository Methods

Advanced Custom Methods

Query Builder Integration

Transaction Support

Best Practices

Complete Example

Next Steps