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

Key features of session management:

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

​Repositories

​What is a Repository?

​Core Concepts

​Repository Hierarchy

​BaseRepository

​SQLModelRepository

​Session Management

​Implementation

​Basic Repository Implementation

​Advanced Query Patterns

​Database Initialization

​Integration with Other Components

​Service Integration

​Configuration Integration

​Flavor Integration

​Best Practices

​Do’s

​Don’ts