add-feature

name	add-feature
description	Step-by-step guide for adding a complete new feature to this FastAPI template. Use when user asks to "add a feature", "create a new module", "implement X", or when a task touches all layers (route + service + repository + model).

Add features following the established layered architecture. Every feature touches every layer.

Step-by-Step Process

1. Define the SQLAlchemy Model

Create in app/models/ (e.g., app/models/product.py):

from sqlalchemy import Column, Integer, String, DateTime, func
from app.database.db import Base

class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255), nullable=False, unique=True)
    created_at = Column(DateTime, server_default=func.now())

Export from app/models/db_model.py:

from app.models.product import Product

2. Define Pydantic Schemas

Create app/schema/product.py:

from pydantic import BaseModel, Field
from app.core.validation import SafeStr

class ProductCreate(BaseModel):
    name: SafeStr = Field(..., min_length=1, max_length=255)

class ProductResponse(BaseModel):
    id: int
    name: str
    model_config = {"from_attributes": True}

3. Create Repository

Create app/repositories/product/product.py:

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.logger import LoggedRepository
from app.models.db_model import Product
from app.schema.product import ProductCreate

class ProductRepository(LoggedRepository):
    def __init__(self, db: AsyncSession):
        self.db = db

    async def create(self, data: ProductCreate) -> Product:
        product = Product(**data.model_dump())
        self.db.add(product)
        await self.db.commit()
        await self.db.refresh(product)
        return product

    async def get_all(self, skip: int = 0, limit: int = 20) -> list[Product]:
        result = await self.db.execute(
            select(Product).offset(skip).limit(limit)
        )
        return list(result.scalars().all())

    async def get_by_id(self, product_id: int) -> Product | None:
        result = await self.db.execute(select(Product).where(Product.id == product_id))
        return result.scalar_one_or_none()

4. Create Service

Create app/services/product/service.py:

from app.core.logger import LoggedService
from app.error.custom_exception import AppException
from app.repositories.product.product import ProductRepository
from app.schema.product import ProductCreate, ProductResponse
from app.models.db_model import Product

class ProductService(LoggedService):
    def __init__(self, repo: ProductRepository):
        self.repo = repo

    async def create_product(self, data: ProductCreate) -> ProductResponse:
        existing = await self.repo.get_by_name(data.name)
        if existing:
            raise AppException(
                f"Product '{data.name}' already exists",
                error_code="product_exists"
            )
        product = await self.repo.create(data)
        return ProductResponse.model_validate(product)

    async def list_products(self, skip: int = 0, limit: int = 20) -> list[ProductResponse]:
        products = await self.repo.get_all(skip=skip, limit=limit)
        return [ProductResponse.model_validate(p) for p in products]

5. Wire Dependencies

Add to app/core/dependency_factory.py:

def get_product_repository(db: AsyncSession = Depends(get_db)) -> ProductRepository:
    return ProductRepository(db)

def get_product_service(
    repo: ProductRepository = Depends(get_product_repository),
) -> ProductService:
    return ProductService(repo)

6. Create Route

Create app/api/v1/routes/product.py:

from typing import Annotated
from fastapi import APIRouter, Depends, status
from app.core.dependency_factory import get_product_service
from app.core.logger import log_function
from app.schema.product import ProductCreate, ProductResponse
from app.services.product.service import ProductService
from app.core.dependencies import role_required

router = APIRouter(tags=["Products"])

@router.post("/", response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
@log_function
async def create_product(
    data: ProductCreate,
    service: Annotated[ProductService, Depends(get_product_service)],
    _: None = Depends(role_required(["admin"])),
):
    return await service.create_product(data)

@router.get("/", response_model=list[ProductResponse])
@log_function
async def list_products(
    service: Annotated[ProductService, Depends(get_product_service)],
    skip: int = 0,
    limit: int = 20,
):
    return await service.list_products(skip=skip, limit=limit)

7. Register Router

In app/main.py:

from app.api.v1.routes import product
app.include_router(product.router, prefix="/api/v1/products")

8. Create Migration

alembic revision --autogenerate -m "add products table"
alembic upgrade head

Checklist

• Model in app/models/ + exported from db_model.py
• Schema in app/schema/ with model_config = {"from_attributes": True}
• Repository extends LoggedRepository
• Service extends LoggedService, validates + raises exceptions
• Dependencies wired in dependency_factory.py
• Router registered in app/main.py
• Migration created and applied
• Tests written (see write-tests skill)