Skip to main content

Creating Your First Flavor

This guide will walk you through creating a custom flavor from scratch. We’ll build a simple “Task” flavor that demonstrates all the core concepts using the actual patterns from the Tasteful codebase.

Planning Your Flavor

Before coding, consider:
  • Domain responsibility: What specific functionality will this flavor handle?
  • API endpoints: What HTTP routes do you need?
  • Services: What business logic components are required?
  • Dependencies: What other services or data stores do you need?
For our Task flavor, we’ll need:
  • CRUD operations for task items
  • Task service for business logic
  • Repository for data persistence
  • Configuration for database connection

Step 1: Create the Basic Structure

Create the flavor directory structure (following the same pattern as the built-in HealthFlavor):
flavors/
└── task/
    ├── __init__.py
    ├── flavor.py       # Main flavor definition
    ├── config.py       # Configuration class
    ├── controller.py   # HTTP endpoints and routing
    ├── service.py      # Business logic service
    └── repository.py   # Data access layer

Step 2: Create Configuration

Start with the configuration class in config.py. The BaseConfig class provides automatic environment variable loading and validation:
# flavors/task/config.py
from tasteful.base_flavor import BaseConfig
from pydantic import Field

class TaskConfig(BaseConfig):
    """Configuration for Task flavor."""
    
    database_url: str = "sqlite:///tasks.db"
    max_tasks_per_user: int = Field(default=100, ge=1, le=1000)
    enable_notifications: bool = True
    
    # Environment variables will automatically map:
    # DATABASE_URL -> database_url
    # MAX_TASKS_PER_USER -> max_tasks_per_user
    # ENABLE_NOTIFICATIONS -> enable_notifications
For comprehensive configuration management patterns, see the Configuration Management Guide.

Step 3: Create Repository

Create the repository for data access in repository.py:
# flavors/task/repository.py
from contextlib import contextmanager
from tasteful.repositories import SQLModelRepository
from .config import TaskConfig

class TaskRepository(SQLModelRepository):
    """Repository for task data operations."""
    
    def __init__(self, task_config: TaskConfig) -> None:
        super().__init__(database_url=task_config.database_url, echo=False)
        self.task_config = task_config
    
    def test_connection(self) -> dict:
        """Test database connection and return status."""
        try:
            with self.get_session():
                # Mask password in URL if present
                masked_url = str(self._engine.url)
                if self._engine.url.password:
                    masked_url = str(self._engine.url).replace(
                        self._engine.url.password, "***"
                    )
                
                return {
                    "status": "connected",
                    "database_url": masked_url,
                }
        except Exception as e:
            return {
                "status": "error", 
                "error": str(e),
            }
    
    def get_task_count(self) -> int:
        """Get total number of tasks."""
        # In a real implementation, you'd query the database
        return 42
    
    def create_task(self, title: str, description: str = None) -> dict:
        """Create a new task."""
        # In a real implementation, you'd save to database
        return {
            "id": 1,
            "title": title,
            "description": description,
            "completed": False
        }

Step 4: Implement the Service

Create the business logic in service.py:
# flavors/task/service.py
from tasteful.base_flavor import BaseService
from .repository import TaskRepository
from .config import TaskConfig

class TaskService(BaseService):
    """Service for task business logic."""
    
    def __init__(self, task_repository: TaskRepository, task_config: TaskConfig) -> None:
        super().__init__()
        self.task_repository = task_repository
        self.task_config = task_config
    
    def get_service_info(self) -> dict:
        """Get service information."""
        return {
            "service": "task-service",
            "version": "1.0.0",
            "max_tasks": self.task_config.max_tasks_per_user
        }
    
    def check_database_status(self) -> dict:
        """Check database connection status."""
        return self.task_repository.test_connection()
    
    def get_task_stats(self) -> dict:
        """Get task statistics."""
        return {
            "total_tasks": self.task_repository.get_task_count(),
            "max_tasks_per_user": self.task_config.max_tasks_per_user,
            "notifications_enabled": self.task_config.enable_notifications
        }
    
    def create_task(self, title: str, description: str = None) -> dict:
        """Create a new task."""
        current_count = self.task_repository.get_task_count()
        if current_count >= self.task_config.max_tasks_per_user:
            raise ValueError("Maximum number of tasks reached")
        
        return self.task_repository.create_task(title, description)

Step 5: Create the Controller

Create the HTTP controller in controller.py:
# flavors/task/controller.py
from fastapi import HTTPException
from tasteful.base_flavor import BaseController
from tasteful.decorators.controller import Get, Post
from .service import TaskService

class TaskController(BaseController):
    """Controller for task HTTP endpoints."""
    
    def __init__(self, task_service: TaskService) -> None:
        super().__init__(prefix="/tasks", tags=["tasks"])
        self.task_service = task_service
    
    @Get("/")
    async def get_service_info(self) -> dict:
        """Get task service information."""
        return self.task_service.get_service_info()
    
    @Get("/status")
    async def get_status(self) -> dict:
        """Get task service status including database connection."""
        return self.task_service.check_database_status()
    
    @Get("/stats")
    async def get_task_stats(self) -> dict:
        """Get task statistics."""
        return self.task_service.get_task_stats()
    
    @Post("/")
    async def create_task(self, title: str, description: str = None) -> dict:
        """Create a new task."""
        try:
            return self.task_service.create_task(title, description)
        except ValueError as e:
            raise HTTPException(status_code=400, detail=str(e))

Step 6: Create the Flavor

Now implement the main flavor in flavor.py:
# flavors/task/flavor.py
from tasteful.base_flavor import BaseFlavor
from .config import TaskConfig
from .controller import TaskController
from .repository import TaskRepository
from .service import TaskService

class TaskFlavor(BaseFlavor):
    """Task flavor with database connectivity and business logic."""
    
    def __init__(self):
        super().__init__(
            controller=TaskController,
            services=[TaskService],
            repositories=[TaskRepository],
            config=TaskConfig,
        )

Step 7: Export the Flavor

Make your flavor importable by updating __init__.py:
# flavors/task/__init__.py
from .config import TaskConfig
from .controller import TaskController
from .flavor import TaskFlavor
from .repository import TaskRepository
from .service import TaskService

__all__ = [
    "TaskFlavor",
    "TaskController", 
    "TaskService",
    "TaskRepository",
    "TaskConfig"
]

Step 8: Register with Your Application

Add the flavor to your main application:
# main.py
from tasteful import TastefulApp
from tasteful.flavors import HealthFlavor
from flavors.task import TaskFlavor

app = TastefulApp(
    title="My Task Application",
    version="1.0.0",
    flavors=[
        HealthFlavor,
        TaskFlavor
    ]
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app.app, host="0.0.0.0", port=8000)

Step 9: Test Your Flavor

Run your application and test the endpoints:
# Start the application
python main.py

# Test in another terminal
curl http://localhost:8000/tasks/

# Get service status
curl http://localhost:8000/tasks/status

# Get task statistics
curl http://localhost:8000/tasks/stats

# Create a task
curl -X POST "http://localhost:8000/tasks/?title=Learn%20Tasteful&description=Complete%20the%20tutorial"
Visit http://localhost:8000/docs to see your API documentation automatically generated by FastAPI.

Advanced Features

Complex Dependency Injection

You can create more complex dependency graphs like in the main.py example:
# Multiple services with dependencies
class NotificationService(BaseService):
    def __init__(self, task_config: TaskConfig):
        super().__init__()
        self.config = task_config
    
    def send_notification(self, message: str) -> bool:
        if self.config.enable_notifications:
            # Send notification logic
            return True
        return False

class TaskService(BaseService):
    def __init__(
        self, 
        task_repository: TaskRepository, 
        notification_service: NotificationService,
        task_config: TaskConfig
    ):
        super().__init__()
        self.task_repository = task_repository
        self.notification_service = notification_service
        self.task_config = task_config
    
    def create_task_with_notification(self, title: str) -> dict:
        task = self.task_repository.create_task(title)
        self.notification_service.send_notification(f"Task created: {title}")
        return task

# Update your flavor
class TaskFlavor(BaseFlavor):
    def __init__(self):
        super().__init__(
            controller=TaskController,
            services=[TaskService, NotificationService],  # Multiple services
            repositories=[TaskRepository],
            config=TaskConfig,
        )

Environment-Specific Configuration

Use environment variables and validation in your config:
from pydantic import Field, validator

class TaskConfig(BaseConfig):
    """Configuration for Task flavor."""
    
    database_url: str = Field(
        default="sqlite:///tasks.db",
        description="Database connection URL"
    )
    max_tasks_per_user: int = Field(
        default=100, 
        ge=1, 
        le=1000,
        description="Maximum tasks per user"
    )
    enable_notifications: bool = True
    
    # Environment-specific settings
    debug_mode: bool = False
    log_level: str = Field(default="INFO", regex="^(DEBUG|INFO|WARNING|ERROR)$")
    external_api_key: str = Field(default="", min_length=0)
    
    @validator('external_api_key')
    def validate_api_key(cls, v):
        if v and len(v) < 10:
            raise ValueError('API key must be at least 10 characters')
        return v
Create a .env file for local development:
# .env
DATABASE_URL=postgresql://user:pass@localhost/tasks
MAX_TASKS_PER_USER=500
DEBUG_MODE=true
LOG_LEVEL=DEBUG
EXTERNAL_API_KEY=your-api-key-here
For advanced configuration patterns including nested configuration, validation, and environment-specific setups, see the Configuration Management Guide.

Best Practices

1. Separation of Concerns

  • Config: Configuration and settings
  • Services: Business logic and orchestration
  • Repositories: Data access and persistence
  • Controllers: HTTP layer and request/response handling
  • Flavors: Component composition and dependency wiring

2. Error Handling

Use FastAPI’s HTTPException in controllers for consistent error responses:
from fastapi import HTTPException

class TaskController(BaseController):
    def __init__(self, task_service: TaskService):
        super().__init__(prefix="/tasks", tags=["tasks"])
        self.task_service = task_service

    @Post("/")
    async def create_task(self, title: str, description: str = None) -> dict:
        try:
            return self.task_service.create_task(title, description)
        except ValueError as e:
            raise HTTPException(status_code=400, detail=str(e))
        except Exception as e:
            raise HTTPException(status_code=500, detail="Internal server error")

3. Async Operations

Use async/await for I/O operations in controllers:
class TaskController(BaseController):
    def __init__(self, task_service: TaskService):
        super().__init__(prefix="/tasks", tags=["tasks"])
        self.task_service = task_service

    @Get("/status")
    async def get_status(self) -> dict:
        # Async database call
        status = await self.task_service.check_database_status_async()
        return status

4. Documentation

Add comprehensive docstrings to controllers and services:
class TaskController(BaseController):
    def __init__(self, task_service: TaskService):
        super().__init__(prefix="/tasks", tags=["tasks"])
        self.task_service = task_service

    @Post("/")
    async def create_task(self, title: str, description: str = None) -> dict:
        """
        Create a new task item.
        
        Args:
            title: The task title (required)
            description: Optional task description
            
        Returns:
            The created task with generated ID
            
        Raises:
            HTTPException: 400 if validation fails or max tasks reached
        """
        try:
            return self.task_service.create_task(title, description)
        except ValueError as e:
            raise HTTPException(status_code=400, detail=str(e))

Next Steps

I