fastapi-templates

• 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

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

Pattern 1: Complete FastAPI Application

```python

# 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 Config:

env_file = ".env"

@lru_cache()

def get_settings() -> Settings:

return Settings()

# core/database.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import sessionmaker

from app.core.config import get_settings

settings = get_settings()

engine = create_async_engine(

settings.DATABASE_URL,

echo=True,

future=True

)

AsyncSessionLocal = sessionmaker(

engine,

class_=AsyncSession,

expire_on_commit=False

)

Base = declarative_base()

async def get_db() -> AsyncSession:

"""Dependency for database session."""

async with AsyncSessionLocal() as session:

try:

yield session

await session.commit()

except Exception:

await session.rollback()

raise

finally:

await session.close()

```

Pattern 2: CRUD Repository Pattern

```python

# 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

) -> List[ModelType]:

"""Get multiple records."""

result = await db.execute(

select(self.model).offset(skip).limit(limit)

)

return result.scalars().all()

async def create(

self,

db: AsyncSession,

obj_in: CreateSchemaType

) -> ModelType:

"""Create new record."""

db_obj = self.model(**obj_in.dict())

db.add(db_obj)

await db.flush()

await db.refresh(db_obj)

return db_obj

async def update(

self,

db: AsyncSession,

db_obj: ModelType,

obj_in: UpdateSchemaType

) -> ModelType:

"""Update record."""

update_data = obj_in.dict(exclude_unset=True)

for field, value in update_data.items():

setattr(db_obj, field, value)

await db.flush()

await db.refresh(db_obj)

return db_obj

async def delete(self, db: AsyncSession, id: int) -> bool:

"""Delete record."""

obj = await self.get(db, id)

if obj:

await db.delete(obj)

return True

return False

# repositories/user_repository.py

from app.repositories.base_repository import BaseRepository

from app.models.user import User

from app.schemas.user import UserCreate, UserUpdate

class UserRepository(BaseRepository[User, UserCreate, UserUpdate]):

"""User-specific repository."""

async def get_by_email(self, db: AsyncSession, email: str) -> Optional[User]:

"""Get user by email."""

result = await db.execute(

select(User).where(User.email == email)

)

return result.scalars().first()

async def is_active(self, db: AsyncSession, user_id: int) -> bool:

"""Check if user is active."""

user = await self.get(db, user_id)

return user.is_active if user else False

user_repository = UserRepository(User)

```

Pattern 3: Service Layer

```python

# 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: str,

password: str

) -> Optional[User]:

"""Authenticate user."""

user = await self.repository.get_by_email(db, email)

if not user:

return None

if not verify_password(password, user.hashed_password):

return None

return user

async def update_user(

self,

db: AsyncSession,

user_id: int,

user_in: UserUpdate

) -> Optional[User]:

"""Update user."""

user = await self.repository.get(db, user_id)

if not user:

return None

if user_in.password:

user_in_dict = user_in.dict(exclude_unset=True)

user_in_dict["hashed_password"] = get_password_hash(

user_in_dict.pop("password")

)

user_in = UserUpdate(**user_in_dict)

return await self.repository.update(db, user, user_in)

user_service = UserService()

```

Pattern 4: API Endpoints with Dependencies

```python

# 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)

async def read_user(

user_id: int,

db: AsyncSession = Depends(get_db),

current_user: User = Depends(get_current_user)

):

"""Get user by ID."""

user = await user_service.repository.get(db, user_id)

if not user:

raise HTTPException(status_code=404, detail="User not found")

return user

@router.patch("/{user_id}", response_model=User)

async def update_user(

user_id: int,

user_in: UserUpdate,

db: AsyncSession = Depends(get_db),

current_user: User = Depends(get_current_user)

):

"""Update user."""

if current_user.id != user_id:

raise HTTPException(status_code=403, detail="Not authorized")

user = await user_service.update_user(db, user_id, user_in)

if not user:

raise HTTPException(status_code=404, detail="User not found")

return user

@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)

async def delete_user(

user_id: int,

db: AsyncSession = Depends(get_db),

current_user: User = Depends(get_current_user)

):

"""Delete user."""

if current_user.id != user_id:

raise HTTPException(status_code=403, detail="Not authorized")

deleted = await user_service.repository.delete(db, user_id)

if not deleted:

raise HTTPException(status_code=404, detail="User not found")

```

Pattern 5: Authentication & Authorization

```python

# 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

from fastapi import Depends, HTTPException, status

from fastapi.security import OAuth2PasswordBearer

from jose import JWTError, jwt

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db

from app.core.security import ALGORITHM

from app.core.config import get_settings

from app.repositories.user_repository import user_repository

oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")

async def get_current_user(

db: AsyncSession = Depends(get_db),

token: str = Depends(oauth2_scheme)

):

"""Get current authenticated user."""

credentials_exception = HTTPException(

status_code=status.HTTP_401_UNAUTHORIZED,

detail="Could not validate credentials",

headers={"WWW-Authenticate": "Bearer"},

)

try:

payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])

user_id: int = payload.get("sub")

if user_id is None:

raise credentials_exception

except JWTError:

raise credentials_exception

user = await user_repository.get(db, user_id)

if user is None:

raise credentials_exception

return user

```

```python

# 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="http://test") as client:

yield client

# tests/test_users.py

import pytest

@pytest.mark.asyncio

async def test_create_user(client):

response = await client.post(

"/api/v1/users/",

json={

"email": "test@example.com",

"password": "testpass123",

"name": "Test User"

}

)

assert response.status_code == 201

data = response.json()

assert data["email"] == "test@example.com"

assert "id" in data

```

• 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

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

• 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