Skip to main content

Creating Your First Feature

This guide walks you through creating a complete backend feature from scratch: a task management system (aka the good old TODO list). You'll learn how to create database models, API endpoints and test everything together.

What We'll Build​

A simple task management feature with:

  • Task model with title, description, and completion status
  • Full CRUD API endpoints
  • User authentication
  • Database migration

Prerequisites​

  • Starter app is set up and running
  • Backend development environment is configured
  • Basic understanding of Python and FastAPI

Step 1: Create the Database Model​

First, let's define our task model in common/src/common/models/db_models.py:

# Add these imports at the top if not already present
from datetime import datetime
from uuid import UUID
from sqlmodel import Field, Relationship

# Add this new model at the end of the file
class TaskBase(ModelWithIdAndTimestamps):
    title: str = Field(max_length=200)
    description: str | None = Field(default=None, max_length=1000)
    completed: bool = Field(default=False)
    due_date: datetime | None = Field(default=None)
    user_id: UUID = Field(foreign_key="user.id", index=True)

class Task(TaskBase, table=True):
    user: User = Relationship(back_populates="tasks")

Update the User model to include the relationship:

# In the User class, add this line:
class User(UserBase, table=True):
    group: Group | None = Relationship(back_populates="users")
    tasks: list[Task] = Relationship(back_populates="user")  # Add this line
Relationships and Performance

The Relationship fields define how models are connected. Since we're using AsyncSession, lazy loading is not supported - accessing a relationship without eager loading will raise an error.

You have two options:

  1. Don't access the relationship if you don't need it
  2. Eager load it using joinedload when you do need it
# Efficient: Single query with JOIN
from sqlalchemy.orm import joinedload

result = await db.exec(
    select(Task).options(joinedload(Task.user))
)
tasks = result.unique().all()

# Inefficient: N+1 queries (1 for tasks + N for users)
tasks = await db.exec(select(Task))
for task in tasks.all():
    print(task.user.name)  # Each access triggers a new query!

Learn more in the Database operations guide.

Step 2: Create API Models​

Add the API models in apps/backend/src/models/api_models.py:

# Add at the end of the file
CreateTaskDTO = create_create_model(TaskBase, "CreateTask")
UpdateTaskDTO = create_update_model(TaskBase, "UpdateTask")

class TaskDTO(BaseDTO, TaskBase):
    pass

class TaskWithUserDTO(TaskDTO):
    user: UserDTO | None = None
tip

You can also check the Pydantic Models guide.

Step 3: Generate Database Migration​

Create and apply the database migration:

# From project root
uvx invoke create-db-migration "Add tasks table"
uvx invoke update-db
tip

You can also check the Database Migration guide.

Step 4: Create the Router​

Create apps/backend/src/routers/task.py:

from typing import List
from uuid import UUID

from fastapi import HTTPException, Depends
from sqlalchemy.orm import joinedload
from sqlmodel import select

from models.api_models import CreateTaskDTO, UpdateTaskDTO, TaskDTO, TaskWithUserDTO
from common.models.db_models import Task, User
from routers.router_utils import CRUDBaseRouter
from common.services.database_service import DatabaseDep
from utils.auth import get_current_user

# CRUD router with automatic endpoints
_task_crud: CRUDBaseRouter = CRUDBaseRouter(
    model=Task,
    read_schema=TaskDTO,
    create_schema=CreateTaskDTO,
    update_schema=UpdateTaskDTO,
    prefix="/tasks",
    tags=["Tasks"],
    dependencies=[],
)

router = _task_crud.get_router()

# Custom endpoints for user-specific operations
@router.get("/my-tasks", response_model=List[TaskDTO])
async def get_my_tasks(
    current_user: User | None = Depends(get_current_user),
    db: DatabaseDep,  # FastAPI dependency injection for database session
) -> list[TaskDTO]:
    """Get all tasks for the authenticated user"""
    if not current_user:
        raise HTTPException(status_code=401, detail="Not authenticated")

    result = await db.exec(
        select(Task).where(Task.user_id == current_user.id)
    )
    tasks = result.all()
    return [TaskDTO.model_validate(task) for task in tasks]

@router.post("/my-tasks", response_model=TaskDTO)
async def create_my_task(
    task_data: CreateTaskDTO,
    current_user: User | None = Depends(get_current_user),
    db: DatabaseDep,
) -> TaskDTO:
    """Create a new task for the authenticated user"""
    if not current_user:
        raise HTTPException(status_code=401, detail="Not authenticated")

    task = Task.model_validate(task_data)
    task.user_id = current_user.id

    db.add(task)
    await db.commit()
    await db.refresh(task)
    return TaskDTO.model_validate(task)

@router.patch("/{task_id}/toggle", response_model=TaskDTO)
async def toggle_task_completion(
    task_id: UUID,
    current_user: User | None = Depends(get_current_user),
    db: DatabaseDep,
) -> TaskDTO:
    """Toggle the completion status of a task"""
    if not current_user:
        raise HTTPException(status_code=401, detail="Not authenticated")

    result = await db.exec(
        select(Task).where(
            Task.id == task_id,
            Task.user_id == current_user.id
        )
    )
    task = result.first()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    task.completed = not task.completed
    await db.commit()
    await db.refresh(task)
    return TaskDTO.model_validate(task)

# Setup the CRUD routes
_task_crud.setup_routes()
Database Session Management

In this example, we use DatabaseDep which is FastAPI's dependency injection system for database sessions. The framework automatically handles opening and closing the connection.

For operations outside of FastAPI routes (like in Temporal activities or background tasks), use DatabaseService.get_session() instead:

from common.services.database_service import DatabaseService

async def process_tasks():
    async with DatabaseService.get_session() as session:
        result = await session.exec(select(Task))
        return result.all()

Learn more about database operations and query optimization →

Step 5: Register the Router​

Add the router to apps/backend/src/main.py:

# Add this import with the other router imports
from routers import task

# Add this line with the other router inclusions
app.include_router(task.router)

Step 6: Test Your API​

Start the backend server:

# From the project root
uvx invoke backend

Navigate to http://localhost:8000/docs to see your new API endpoints in the interactive documentation.

Test the Endpoints​

  1. Create a task: POST to /tasks/my-tasks

    {
      "title": "Learn FastAPI",
      "description": "Complete the task management tutorial"
    }

  2. Get your tasks: GET /tasks/my-tasks

  3. Toggle completion: PATCH /tasks/{task_id}/toggle

Step 7: Verify Everything Works​

Test your complete feature:

# Check database contains your task
# From project root
uv run python -c "
from sqlmodel import create_engine, Session, select
from common.models.db_models import Task
from common.services.config_service import ConfigService

engine = create_engine(ConfigService.get_database_url())

with Session(engine) as session:
    tasks = session.exec(select(Task)).all()
    print(f'Found {len(tasks)} tasks')
    for task in tasks:
        print(f'- {task.title}: {\"✓\" if task.completed else \"○\"}')
"

What You've Learned​

✅ Database Models: Created SQLModel classes with relationships ✅ API Models: Created DTOs for request/response validation ✅ Database Migrations: Generated and applied schema changes ✅ API Routing: Created both CRUD and custom endpoints ✅ Authentication: Protected routes with user context

🚀 Next Steps​

Your task management API is complete! Now build the frontend interface that connects to it:

Frontend: Your First Component →

Learn how to create a React interface that uses your new API with type-safe calls, state management, and a polished UI.