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

Testing Your Flavor

Learn how to write comprehensive tests for your flavors

Adding Services

Explore advanced service patterns and dependency injection

Database Integration

Connect your flavors to databases with repositories

Authentication

Secure your flavors with authentication and authorization