Skip to main content

​
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
⌘I