fastapi-templates

name	fastapi-templates
description	Create production-ready FastAPI projects with async patterns, dependency injection, and comprehensive error handling. Use when building new FastAPI applications or setting up backend API projects. Use when this capability is needed.
metadata	{"author":"AI-Foundry-Core"}

FastAPI Project Templates

Production-ready FastAPI project structures with async patterns, dependency injection, middleware, and best practices for building high-performance APIs.

When to Use This Skill

• Starting new FastAPI projects from scratch
• Implementing async REST APIs with Python
• Building high-performance web services and microservices
• Creating async applications with PostgreSQL, MongoDB
• Setting up API projects with proper structure and testing

Core Concepts

1. Project Structure

Recommended Layout:

app/
├── api/                    # API routes
│   ├── v1/
│   │   ├── endpoints/
│   │   │   ├── users.py
│   │   │   ├── auth.py
│   │   │   └── items.py
│   │   └── router.py
│   └── dependencies.py     # Shared dependencies
├── core/                   # Core configuration
│   ├── config.py
│   ├── security.py
│   └── database.py
├── models/                 # Database models
│   ├── user.py
│   └── item.py
├── schemas/                # Pydantic schemas
│   ├── user.py
│   └── item.py
├── services/               # Business logic
│   ├── user_service.py
│   └── auth_service.py
├── repositories/           # Data access
│   ├── user_repository.py
│   └── item_repository.py
└── main.py                 # Application entry

2. Dependency Injection

FastAPI's built-in DI system using Depends:

• Database session management
• Authentication/authorization
• Shared business logic
• Configuration injection

3. Async Patterns

Proper async/await usage:

• Async route handlers
• Async database operations
• Async background tasks
• Async middleware

Implementation Patterns

Pattern 1: Complete FastAPI Application

# main.py
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan events."""
    # Startup
    await database.connect()
    yield
    # Shutdown
    await database.disconnect()

app = FastAPI(
    title="API Template",
    version="1.0.0",
    lifespan=lifespan
)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Include routers
from app.api.v1.router import api_router
app.include_router(api_router, prefix="/api/v1")

# core/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    """Application settings."""
    DATABASE_URL: str
    SECRET_KEY: str
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    API_V1_STR: str = "/api/v1"

    class :
        env_file = 


 () -> Settings:
     Settings()


 sqlalchemy.ext.asyncio  create_async_engine, AsyncSession
 sqlalchemy.ext.declarative  declarative_base
 sqlalchemy.orm  sessionmaker
 app.core.config  get_settings

settings = get_settings()

engine = create_async_engine(
    settings.DATABASE_URL,
    echo=,
    future=
)

AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=
)

Base = declarative_base()

  () -> AsyncSession:
    
      AsyncSessionLocal()  session:
        :
             session
             session.commit()
         Exception:
             session.rollback()
            
        :
             session.close()

Pattern 2: CRUD Repository Pattern

# repositories/base_repository.py
from typing import Generic, TypeVar, Type, Optional, List
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from pydantic import BaseModel

ModelType = TypeVar("ModelType")
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)

class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    """Base repository for CRUD operations."""

    def __init__(self, model: Type[ModelType]):
        self.model = model

    async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]:
        """Get by ID."""
        result = await db.execute(
            select(self.model).where(self.model.id == id)
        )
        return result.scalars().first()

    async def get_multi(
        self,
        db: AsyncSession,
        skip: int = 0,
        limit: int = 100
    ) -> [ModelType]:
        
        result =  db.execute(
            select(.model).offset(skip).limit(limit)
        )
         result.scalars().()

      () -> ModelType:
        
        db_obj = .model(**obj_in.())
        db.add(db_obj)
         db.flush()
         db.refresh(db_obj)
         db_obj

      () -> ModelType:
        
        update_data = obj_in.(exclude_unset=)
         field, value  update_data.items():
            (db_obj, field, value)
         db.flush()
         db.refresh(db_obj)
         db_obj

      () -> :
        
        obj =  .get(db, )
         obj:
             db.delete(obj)
             
         


 app.repositories.base_repository  BaseRepository
 app.models.user  User
 app.schemas.user  UserCreate, UserUpdate

 (BaseRepository[User, UserCreate, UserUpdate]):
    

      () -> [User]:
        
        result =  db.execute(
            select(User).where(User.email == email)
        )
         result.scalars().first()

      () -> :
        
        user =  .get(db, user_id)
         user.is_active  user  

user_repository = UserRepository(User)

Pattern 3: Service Layer

# services/user_service.py
from typing import Optional
from sqlalchemy.ext.asyncio import AsyncSession
from app.repositories.user_repository import user_repository
from app.schemas.user import UserCreate, UserUpdate, User
from app.core.security import get_password_hash, verify_password

class UserService:
    """Business logic for users."""

    def __init__(self):
        self.repository = user_repository

    async def create_user(
        self,
        db: AsyncSession,
        user_in: UserCreate
    ) -> User:
        """Create new user with hashed password."""
        # Check if email exists
        existing = await self.repository.get_by_email(db, user_in.email)
        if existing:
            raise ValueError("Email already registered")

        # Hash password
        user_in_dict = user_in.dict()
        user_in_dict["hashed_password"] = get_password_hash(user_in_dict.pop("password"))

        # Create user
        user = await self.repository.create(db, UserCreate(**user_in_dict))
        return user

    async def authenticate(
        self,
        db: AsyncSession,
        email: ,
        password: 
    ) -> [User]:
        
        user =  .repository.get_by_email(db, email)
          user:
             
          verify_password(password, user.hashed_password):
             
         user

      () -> [User]:
        
        user =  .repository.get(db, user_id)
          user:
             

         user_in.password:
            user_in_dict = user_in.(exclude_unset=)
            user_in_dict[] = get_password_hash(
                user_in_dict.pop()
            )
            user_in = UserUpdate(**user_in_dict)

          .repository.update(db, user, user_in)

user_service = UserService()

Pattern 4: API Endpoints with Dependencies

# api/v1/endpoints/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List

from app.core.database import get_db
from app.schemas.user import User, UserCreate, UserUpdate
from app.services.user_service import user_service
from app.api.dependencies import get_current_user

router = APIRouter()

@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    db: AsyncSession = Depends(get_db)
):
    """Create new user."""
    try:
        user = await user_service.create_user(db, user_in)
        return user
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.get("/me", response_model=User)
async def read_current_user(
    current_user: User = Depends(get_current_user)
):
    """Get current user."""
    return current_user

@router.get("/{user_id}", response_model=User)
  ():
    
    user =  user_service.repository.get(db, user_id)
      user:
         HTTPException(status_code=, detail=)
     user


  ():
    
     current_user. != user_id:
         HTTPException(status_code=, detail=)

    user =  user_service.update_user(db, user_id, user_in)
      user:
         HTTPException(status_code=, detail=)
     user


  ():
    
     current_user. != user_id:
         HTTPException(status_code=, detail=)

    deleted =  user_service.repository.delete(db, user_id)
      deleted:
         HTTPException(status_code=, detail=)

Pattern 5: Authentication & Authorization

# core/security.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import get_settings

settings = get_settings()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """Create JWT access token."""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify password against hash."""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """Hash password."""
    return pwd_context.hash(password)

# api/dependencies.py
 fastapi  Depends, HTTPException, status
 fastapi.security  OAuth2PasswordBearer
 jose  JWTError, jwt
 sqlalchemy.ext.asyncio  AsyncSession

 app.core.database  get_db
 app.core.security  ALGORITHM
 app.core.config  get_settings
 app.repositories.user_repository  user_repository

oauth2_scheme = OAuth2PasswordBearer(tokenUrl=)

  ():
    
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail=,
        headers={: },
    )

    :
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
        user_id:  = payload.get()
         user_id  :
             credentials_exception
     JWTError:
         credentials_exception

    user =  user_repository.get(db, user_id)
     user  :
         credentials_exception

     user

Testing

# tests/conftest.py
import pytest
import asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.core.database import get_db, Base

TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture
async def db_session():
    engine = create_async_engine(TEST_DATABASE_URL, echo=True)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    AsyncSessionLocal = sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False
    )

    async with AsyncSessionLocal() as session:
        yield session

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db

    async with AsyncClient(app=app, base_url=)  client:
         client


 pytest


  ():
    response =  client.post(
        ,
        json={
            : ,
            : ,
            : 
        }
    )
     response.status_code == 
    data = response.json()
     data[] == 
       data

Resources

• references/fastapi-architecture.md: Detailed architecture guide
• references/async-best-practices.md: Async/await patterns
• references/testing-strategies.md: Comprehensive testing guide
• assets/project-template/: Complete FastAPI project
• assets/docker-compose.yml: Development environment setup

Best Practices

1. Async All The Way: Use async for database, external APIs
2. Dependency Injection: Leverage FastAPI's DI system
3. Repository Pattern: Separate data access from business logic
4. Service Layer: Keep business logic out of routes
5. Pydantic Schemas: Strong typing for request/response
6. Error Handling: Consistent error responses
7. Testing: Test all layers independently

Common Pitfalls

• Blocking Code in Async: Using synchronous database drivers
• No Service Layer: Business logic in route handlers
• Missing Type Hints: Loses FastAPI's benefits
• Ignoring Sessions: Not properly managing database sessions
• No Testing: Skipping integration tests
• Tight Coupling: Direct database access in routes

---

Source: AI-Foundry-Core/ril-agents — distributed by TomeVault.