Documentation Index
Fetch the complete documentation index at: https://docs.tasteful.heka.ai/llms.txt
Use this file to discover all available pages before exploring further.
Repositories
Repositories are the data access layer of Tasteful flavors. They provide an abstraction over data storage and retrieval, isolating business logic from database-specific operations. Repositories handle all interactions with databases, external APIs, and other data sources while providing a clean, testable interface for services.
What is a Repository?
A Repository in Tasteful is a class that inherits from BaseRepository or SQLModelRepository and encapsulates data access logic. Repositories are responsible for:
- Data Access Abstraction: Hiding database-specific implementation details from business logic
- CRUD Operations: Providing Create, Read, Update, and Delete functionality
- Query Management: Handling complex queries and data filtering
- Connection Management: Managing database connections and sessions
- Data Mapping: Converting between database records and domain objects
- Transaction Support: Ensuring data consistency through proper transaction handling
Repositories act as the single point of data access for each entity or aggregate, providing a clean interface that services can use without knowing about database specifics.
Core Concepts
Repository Hierarchy
Tasteful provides two base repository classes:
BaseRepository
The foundational repository class that provides basic structure:
from tasteful.repositories import BaseRepository
class BaseRepository:
def __init__(self, database_url: str = None, echo: bool = False):
self.database_url = database_url
self.echo = echo
self._engine = None
SQLModelRepository
The recommended repository class for SQLModel-based applications:
from tasteful.repositories import SQLModelRepository
class UserRepository(SQLModelRepository):
def __init__(self, database_url: str, echo: bool = False):
super().__init__(database_url=database_url, echo=echo)
Session Management
SQLModelRepository provides robust session management through context managers:
class UserRepository(SQLModelRepository):
def get_user_by_id(self, user_id: int) -> Optional[User]:
with self.get_session() as session:
return session.get(User, user_id)
def create_user(self, user_data: dict) -> User:
with self.get_session() as session:
user = User(**user_data)
session.add(user)
session.commit()
session.refresh(user)
return user
- Automatic Rollback: Sessions automatically rollback on exceptions
- Connection Pooling: Efficient database connection management
- Transaction Safety: Proper transaction boundaries for data consistency
Implementation
Basic Repository Implementation
Here’s a complete example of a repository implementation:
from typing import List, Optional
from sqlmodel import select
from tasteful.repositories import SQLModelRepository
class UserRepository(SQLModelRepository):
def __init__(self, database_url: str):
super().__init__(database_url=database_url, echo=False)
def get_by_id(self, user_id: int) -> Optional[User]:
"""Get a user by ID."""
with self.get_session() as session:
return session.get(User, user_id)
def get_by_email(self, email: str) -> Optional[User]:
"""Get a user by email address."""
with self.get_session() as session:
statement = select(User).where(User.email == email)
return session.exec(statement).first()
def get_all(self, skip: int = 0, limit: int = 100) -> List[User]:
"""Get all users with pagination."""
with self.get_session() as session:
statement = select(User).offset(skip).limit(limit)
return list(session.exec(statement).all())
def create(self, user_data: dict) -> User:
"""Create a new user."""
with self.get_session() as session:
user = User(**user_data)
session.add(user)
session.commit()
session.refresh(user)
return user
def update(self, user_id: int, user_data: dict) -> Optional[User]:
"""Update an existing user."""
with self.get_session() as session:
user = session.get(User, user_id)
if not user:
return None
for key, value in user_data.items():
setattr(user, key, value)
session.add(user)
session.commit()
session.refresh(user)
return user
def delete(self, user_id: int) -> bool:
"""Delete a user by ID."""
with self.get_session() as session:
user = session.get(User, user_id)
if not user:
return False
session.delete(user)
session.commit()
return True
Advanced Query Patterns
Repositories can implement complex queries while maintaining clean interfaces:
class UserRepository(SQLModelRepository):
def find_active_users_by_role(self, role: str) -> List[User]:
"""Find active users with a specific role."""
with self.get_session() as session:
statement = select(User).where(
User.is_active == True,
User.role == role
)
return list(session.exec(statement).all())
def get_users_created_after(self, date: datetime) -> List[User]:
"""Get users created after a specific date."""
with self.get_session() as session:
statement = select(User).where(User.created_at > date)
return list(session.exec(statement).all())
def count_users_by_status(self, status: str) -> int:
"""Count users by status."""
with self.get_session() as session:
statement = select(User).where(User.status == status)
return len(list(session.exec(statement).all()))
Database Initialization
Repositories can handle database table creation:
class UserRepository(SQLModelRepository):
def initialize_database(self):
"""Create database tables if they don't exist."""
self.create_db_and_tables()
Integration with Other Components
Service Integration
Repositories are injected into services through dependency injection:
from tasteful.base_flavor import BaseService
class UserService(BaseService):
def __init__(self, user_repository: UserRepository):
super().__init__()
self.user_repository = user_repository
def create_user(self, user_data: dict) -> User:
# Business logic validation
if not user_data.get('email'):
raise ValueError("Email is required")
# Check if user already exists
existing_user = self.user_repository.get_by_email(user_data['email'])
if existing_user:
raise ValueError("User with this email already exists")
# Delegate to repository for data persistence
return self.user_repository.create(user_data)
Configuration Integration
Repositories receive configuration through dependency injection:
class UserRepository(SQLModelRepository):
def __init__(self, app_config: AppConfig):
super().__init__(
database_url=app_config.database_url,
echo=app_config.debug_sql
)
self.app_config = app_config
Flavor Integration
Repositories are registered in flavor containers:
from tasteful.base_flavor import BaseFlavor
class UserFlavor(BaseFlavor):
def __init__(self):
super().__init__()
# Register repository
self.container.repository.add_singleton(
UserRepository,
app_config=self.container.config.app_config
)
# Register service with repository dependency
self.container.service.add_singleton(
UserService,
user_repository=self.container.repository.user_repository
)
Best Practices
Do’s
- Single Responsibility: Each repository should handle one entity or aggregate
- Interface Consistency: Use consistent method naming across repositories (get_by_id, create, update, delete)
- Session Management: Always use the
get_session() context manager for database operations
- Error Handling: Let database exceptions bubble up to services for proper business logic handling
- Query Optimization: Use appropriate indexes and query patterns for performance
- Transaction Boundaries: Keep transactions as short as possible while maintaining consistency
Don’ts
- Business Logic: Don’t put business rules in repositories - they belong in services
- Direct Database Access: Don’t bypass the repository pattern by accessing the database directly from services
- Session Leaks: Don’t store sessions as instance variables - always use context managers
- Complex Joins: Avoid overly complex queries that are hard to maintain - consider breaking them down
- Tight Coupling: Don’t make repositories dependent on other repositories directly
