diegosouzapw This skill guides the creation of a production-ready FastAPI project with all necessary components.
Project Structure
Generate this standardized structure:
project/
├── app/
│ ├── api/ # API modules
│ │ ├── router.py # Central route registration
│ │ └── [module]/ # Feature module (e.g., user, form)
│ │ ├── __init__.py
│ │ ├── models.py # SQLAlchemy models
│ │ ├── schemas.py # Pydantic schemas
│ │ ├── router.py # APIRouter definitions
│ │ ├── views.py # View functions
│ │ └── tasks.py # Celery tasks
│ ├── core/ # Core components
│ │ ├── __init__.py
│ │ ├── exceptions.py # Custom exception handlers
│ │ ├── middleware.py # CORS, compression, rate limiting
│ │ └── oauth2.py # JWT authentication
│ ├── extensions/ # Extensions layer
│ │ ├── __init__.py
│ │ ├── db.py # SQLAlchemy sync/async engines
│ │ ├── cache.py # Redis client
│ │ └── mongo.py # MongoDB Motor client
│ ├── settings/ # Configuration management
│ │ ├── __init__.py # Settings loader
│ │ ├── development.py # Development config
│ │ ├── testing.py # Testing config
│ │ └── production.py # Production config
│ ├── utils/ # Utilities
│ │ ├── __init__.py
│ │ ├── logger.py # Logging setup
│ │ └── helpers.py # Helper functions
├── common/ # Shared modules
│ ├── response.py # Response wrappers (response_ok/response_err)
│ ├── security.py # Password hashing, JWT
│ ├── pagination.py # Custom pagination
│ └── error.py # Custom exceptions
├── tests/ # Test suite
├── alembic/ # Database migrations
│ ├── versions/
│ └── env.py
├── docker/ # Docker configuration
│ ├── Dockerfile
│ └── docker-compose.yml
├── requirements.txt # Python dependencies
├── main.py # Application entry point (root directory)
├── .env # Environment variables
├── .env.example # Environment template
├── .gitignore
└── alembic.ini
Core Dependencies
Use these versions as baseline:
# Core
fastapi>=0.115.0
uvicorn[standard]>=0.30.0
pydantic>=2.5.0
pydantic-settings>=2.1.0
# Database
sqlalchemy>=2.0.25
alembic>=1.13.0
asyncmy>=0.2.10 # Async MySQL driver
redis>=5.0.1
motor>=3.3.0 # MongoDB async
# Auth
PyJWT>=2.10.0
pwdlib[argon2]>=0.3.0
# Rate limiting
slowapi>=0.1.9
# Tasks
celery>=5.3.0
redis>=5.0.1
# Utilities
python-multipart>=0.0.6
python-dotenv>=1.0.0
loguru>=0.7.0
httpx>=0.26.0
limits>=5.6.0
Configuration Management
Use multi-environment settings based on ENV variable:
# app/settings/__init__.py
import os
from typing import Dict
from functools import lru_cache
from dotenv import load_dotenv
load_dotenv('.env')
from app.settings.development import DevelopmentSettings
from app.settings.testing import TestingSettings
from app.settings.production import ProductionSettings
@lru_cache()
def get_settings():
env = os.getenv('ENV', None)
env_config: Dict = {
"development": DevelopmentSettings(),
"testing": TestingSettings(),
"production": ProductionSettings()
}
if env is None or env not in env_config:
raise EnvironmentError("ENV is undefined!")
return env_config[env]
settings = get_settings()
# app/settings/development.py
from pydantic import Field
from typing import List
class DevelopmentSettings:
# App
PROJECT_NAME: str = "FastAPI Project"
VERSION: str = "1.0.0"
ENVIRONMENT: str = "development"
# Database
SQLALCHEMY_DATABASE_ASYNC_URL: str = "mysql+asyncmy://user:pass@localhost:3306/db"
SQLALCHEMY_DATABASE_SYNC_URL: str = "mysql+pymysql://user:pass@localhost:3306/db"
SQLALCHEMY_POOL_SIZE: int = 5
SQLALCHEMY_POOL_RECYCLE: int = 3600
SQLALCHEMY_POOL_PRE_PING: bool = True
SQLALCHEMY_ECHO: bool = False
# Redis
REDIS_URL: str = "redis://localhost:6379/0"
# JWT
JWT_SECRET_KEY: str = "your-secret-key"
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
REFRESH_TOKEN_EXPIRE_DAYS: int = 7
# MongoDB
MONGODB_URL: str = "mongodb://localhost:27017"
MONGODB_DB: str = "app"
# Celery
CELERY_BROKER_URL: str = "redis://localhost:6379/1"
CELERY_RESULT_URL: str = "redis://localhost:6379/2"
# CORS
BACKEND_CORS_ORIGINS: List[str] = ["http://localhost:3000"]
# Swagger
SWAGGER_DOCS_URL: str = "/docs"
SWAGGER_REDOC_URL: str = "/redoc"
OPENAPI_URL: str = "/openapi.json"
Response Wrapper
Use unified response format:
# common/response.py
from typing import Union, List, Dict, Tuple, Optional
from starlette import status
from pydantic.types import JsonValue
from fastapi.responses import JSONResponse
class ErrCode(object):
"""Error codes"""
# Common
SYSTEM_ERROR = (1000, 'System error')
QUERY_NOT_EXISTS = (1001, "Data not exists")
QUERY_HAS_EXISTS = (1002, "Data already exists")
COMMON_INTERNAL_ERR = (1004, 'Server error')
COMMON_PERMISSION_ERR = (1005, 'No access permission')
REQUEST_PARAMS_ERROR = (1006, "Request params error")
DB_INTEGRITY_ERROR = (1007, 'Data conflict')
DB_CONNECTION_ERROR = (1008, 'DB connection failed')
TOO_MANY_REQUEST = (1009, "Too many requests")
# Auth
USER_NOT_EXISTS = (2000, 'User not exists')
USER_HAS_EXISTS = (2001, 'User already exists')
USER_NOT_ACTIVE = (2002, 'User not active')
UNAME_OR_PWD_ERROR = (2003, "Username or password error")
TOKEN_INVALID_ERROR = (2004, 'Token error')
TOKEN_EXPIRED_ERROR = (2005, 'Token expired')
def response_ok(
data: Optional[JsonValue] = None,
msg: str = 'success',
status_code = status.HTTP_200_OK,
**kwargs
) -> JSONResponse:
"""Success response"""
ret = {'errcode': 0, 'errmsg': msg}
if data is not None:
ret['data'] = data
ret.update({k: v for k, v in kwargs.items() if k not in ret})
return JSONResponse(ret, status_code=status_code)
def response_err(
errcode: Tuple[int, str],
errmsg: Optional[str] = None,
detail: Union[List, Dict, None] = None,
status_code = status.HTTP_200_OK
) -> JSONResponse:
"""Error response"""
ret = {"errcode": errcode[0], "errmsg": errcode[1]}
if errmsg is not None:
ret['errmsg'] = errmsg
if detail is not None:
ret['detail'] = detail
return JSONResponse(ret, status_code=status_code)
Custom Pagination
# common/pagination.py
from typing import Any, Optional
from fastapi import Query
from pydantic import BaseModel
class PageNumberPagination(BaseModel):
"""Page number based pagination"""
page: int = Query(default=1, ge=1, description="Page number")
page_size: int = Query(default=10, ge=1, le=100, description="Page size")
total: int = 0
data: list = []
class Config:
arbitrary_types_allowed = True
Database Setup
# app/extensions/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from app.settings import settings
# Sync engine
engine = create_engine(
url=settings.SQLALCHEMY_DATABASE_SYNC_URL,
pool_size=settings.SQLALCHEMY_POOL_SIZE,
pool_recycle=settings.SQLALCHEMY_POOL_RECYCLE,
pool_pre_ping=settings.SQLALCHEMY_POOL_PRE_PING,
echo=settings.SQLALCHEMY_ECHO
)
session = sessionmaker(
engine,
autocommit=False,
autoflush=False,
expire_on_commit=False
)
# Async engine
async_engine = create_async_engine(
url=settings.SQLALCHEMY_DATABASE_ASYNC_URL,
pool_size=settings.SQLALCHEMY_POOL_SIZE,
pool_recycle=settings.SQLALCHEMY_POOL_RECYCLE,
pool_pre_ping=settings.SQLALCHEMY_POOL_PRE_PING,
echo=settings.SQLALCHEMY_ECHO
)
async_session = async_sessionmaker(
bind=async_engine,
expire_on_commit=False,
autocommit=False,
autoflush=False
)
Redis Cache
# app/extensions/cache.py
import redis.asyncio as aioredis
from app.settings import settings
async def get_redis() -> aioredis.Redis:
"""Dependency for Redis connection"""
return aioredis.from_url(settings.REDIS_URL, decode_responses=True)
class RedisCache:
"""Redis cache utility"""
def __init__(self):
self.redis: aioredis.Redis = None
async def connect(self):
self.redis = await get_redis()
async def close(self):
if self.redis:
await self.redis.close()
async def get(self, key: str) -> str:
return await self.redis.get(key)
async def set(self, key: str, value: str, expire: int = 3600):
await self.redis.set(key, value, ex=expire)
async def delete(self, key: str):
await self.redis.delete(key)
MongoDB
# app/extensions/mongo.py
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
from app.settings import settings
class MongoDB:
client: AsyncIOMotorClient = None
db: AsyncIOMotorDatabase = None
mongodb = MongoDB()
async def get_mongo() -> AsyncIOMotorDatabase:
"""Dependency for MongoDB connection"""
return mongodb.db
async def connect_mongo():
"""Connect to MongoDB on startup"""
mongodb.client = AsyncIOMotorClient(settings.MONGODB_URL)
mongodb.db = mongodb.client[settings.MONGODB_DB]
async def close_mongo():
"""Close MongoDB connection on shutdown"""
if mongodb.client:
mongodb.client.close()
# app/extensions/__init__.py
from app.extensions.db import session, async_session
from app.extensions.cache import get_redis, aioredis
from app.extensions.mongo import get_mongo
async def get_db() -> AsyncSession:
async with async_session() as db:
try:
yield db
await db.commit()
except Exception:
await db.rollback()
raise
finally:
await db.close()
Security & Auth
Password Hashing (pwdlib)
# common/security.py
from pwdlib import PasswordHash
password_hash = PasswordHash.recommended()
def set_password(password: str) -> str:
"""Hash password"""
return password_hash.hash(password)
def verify_password(password: str, password_hash_value: str) -> bool:
"""Verify password"""
return password_hash.verify(password, password_hash_value)
JWT Token
# common/security.py (continued)
from datetime import datetime, timedelta
import jwt
from jwt import PyJWTError
from typing import Union, Any, Optional
from app.settings import settings
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.ALGORITHM)
def create_refresh_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, settings.JWT_REFRESH_SECRET_KEY, algorithm=settings.ALGORITHM)
OAuth2 with Scopes
# app/core/oauth2.py
from fastapi import Security, Depends
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from common.response import ErrCode, response_err
from common.security import create_access_token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
async def get_current_active_user(current_user = Security(get_current_user, scopes=['user'])):
return current_user
async def get_current_admin_user(current_user = Security(get_current_user, scopes=['admin'])):
return current_user
Exception Handlers
# app/core/exceptions.py
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from sqlalchemy.exc import IntegrityError, OperationalError
from common.response import ErrCode, response_err
def register_exceptions(app: FastAPI):
@app.exception_handler(Exception)
async def internal_err_handler(request: Request, exc: Exception):
return response_err(ErrCode.COMMON_INTERNAL_ERR, detail=str(exc))
@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
return response_err(ErrCode.REQUEST_PARAMS_ERROR, detail=exc.errors())
@app.exception_handler(IntegrityError)
async def integrity_error_handler(request: Request, exc: IntegrityError):
return response_err(ErrCode.DB_INTEGRITY_ERROR, detail=str(exc.detail))
@app.exception_handler(OperationalError)
async def db_connection_error(request: Request, exc: OperationalError):
return response_err(ErrCode.DB_CONNECTION_ERROR, detail=str(exc))
API Module Template
# app/api/user/views.py
from fastapi import APIRouter, Query, Body, Depends, Security, BackgroundTasks
from sqlalchemy import select
from typing import Optional
from common.response import ErrCode, response_ok, response_err
from common.pagation import PageNumberPagination
from common.security import set_password
from app.extensions import get_db, AsyncSession, get_redis, aioredis
from app.core.oauth2 import get_current_active_user
from app.api.user.models import User
from app.api.user.schemas import UserCreateSchema
from app.api.user.tasks import generate_user_avatar_task
router = APIRouter()
@router.post('/create', summary='Create user')
async def create_user(
background_tasks: BackgroundTasks,
args: UserCreateSchema,
db: AsyncSession = Depends(get_db),
redis: aioredis.Redis = Depends(get_redis),
current_user: User = Security(get_current_active_user, scopes=['admin'])
):
# Check if user exists
existing = await db.scalar(select(User).where(User.username == args.username))
if existing:
return response_err(ErrCode.UNAME_HAS_OCCUPIED)
user = User(
username=args.username,
password_hash=set_password(args.password),
name=args.name or args.username,
is_active=True,
role=args.role
)
db.add(user)
await db.commit()
background_tasks.add_task(generate_user_avatar_task, user.id)
return response_ok(data=user.to_dict(include=['id']))
@router.get('/list', summary='User list')
async def user_list(
username: str = Query(default=None, description='Username'),
paginate: PageNumberPagination = Depends(),
current_user: User = Security(get_current_active_user, scopes=['admin'])
):
query_filter = []
if username:
query_filter.append(User.username.ilike(f"%{username}%"))
result_data = await paginate(User, query_filter)
return response_ok(**result_data)
@router.get('/info', summary='User info')
async def user_info(
current_user: User = Security(get_current_active_user, scopes=['user'])
):
return response_ok(data=current_user.to_dict())
Main Application Entry
# main.py (project root)
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.settings import settings
from app.api.router import register_router
from app.core.exceptions import register_exceptions
from app.core.middleware import register_middleware
@asynccontextmanager
async def lifespan(app: FastAPI):
register_router(app)
yield
app = FastAPI(
title=settings.PROJECT_NAME,
version=settings.VERSION,
lifespan=lifespan,
docs_url=settings.SWAGGER_DOCS_URL,
redoc_url=settings.SWAGGER_REDOC_URL,
openapi_url=settings.OPENAPI_URL,
)
register_middleware(app)
register_exceptions(app)
Docker Configuration
# docker/Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
# docker/docker-compose.yml
version: '3.8'
services:
app:
build: .
ports:
- "8000:8000"
environment:
- ENV=development
- SQLALCHEMY_DATABASE_ASYNC_URL=mysql+asyncmy://user:pass@db:3306/app
- REDIS_URL=redis://redis:6379/0
depends_on:
- db
- redis
volumes:
- .:/app
db:
image: mysql:8
environment:
MYSQL_ROOT_PASSWORD: rootpass
MYSQL_DATABASE: app
command: --default-authentication-plugin=mysql_native_password
redis:
image: redis:7-alpine
volumes:
mysql_data:
redis_data:
Implementation Workflow
When user requests a FastAPI project:
- Gather Requirements: Project name, database, auth requirements
- Generate Structure: Create directory structure
- Create Config: Settings with multi-environment support
- Setup Response: Response wrappers and error codes
- Setup Auth: JWT with pwdlib password hashing
- Add Database: SQLAlchemy with sync/async support
- Add Cache: Redis integration
- Add Tasks: Celery configuration
- Add Docker: Containerization files
- Write README: Documentation
Generate production-grade code with proper error handling, logging, and unified response format.