| name | fastapi-zero-to-hero |
| description | Complete FastAPI API development framework for Python. Provides comprehensive assistance for building APIs with routing, authentication (JWT, OAuth2, Better Auth), Pydantic models, database integration, and deployment using uv package manager. Use when users ask to build FastAPI applications, implement authentication, create API endpoints, or develop backend services in Python. |
FastAPI Zero to Hero - Complete API Development Framework
Overview
This skill provides comprehensive assistance for FastAPI API development in Python, from basic setup to advanced features. It covers routing, authentication, database integration, testing, and deployment patterns using best practices with uv as the package manager.
What This Skill Does
- Creates FastAPI project structures with recommended organization
- Implements API routing with proper error handling
- Sets up authentication systems (JWT, OAuth2, Better Auth)
- Creates Pydantic models for request/response validation
- Configures database integration (SQLAlchemy/async)
- Provides testing and deployment patterns
- Follows FastAPI best practices and security guidelines
- Uses uv package manager for dependency management
What This Skill Does NOT Do
- Create frontend applications (React, Vue, etc.)
- Manage infrastructure (Docker, Kubernetes, cloud deployment)
- Handle specific business logic implementation beyond API patterns
- Provide complete application code without user requirements
Before Implementation
Gather context to ensure successful implementation:
| Source | Gather |
|---|
| Codebase | Existing structure, patterns, conventions to integrate with |
| Conversation | User's specific API requirements, authentication needs, database preferences |
| Skill References | FastAPI documentation patterns, best practices, security guidelines |
| User Guidelines | Project-specific conventions, team standards, deployment requirements |
Ensure all required context is gathered before implementing.
Only ask user for THEIR specific requirements (domain expertise is in this skill).
Required Clarifications
Ask about USER'S context (not domain knowledge):
- API scope: "What specific API endpoints or functionality do you need?"
- Authentication: "Which authentication method do you prefer (JWT, OAuth2, Better Auth)?"
- Database: "Which database are you planning to use (PostgreSQL, MySQL, etc.)?"
- Deployment: "Where do you plan to deploy the API (Docker, cloud, etc.)?"
Workflow
- Set up project structure and dependencies with uv
- Create basic FastAPI application with proper configuration
- Implement authentication system based on requirements
- Design Pydantic models for data validation
- Set up database integration with SQLAlchemy
- Create API routes with proper error handling
- Add testing framework and write tests
- Prepare deployment configuration
Project Setup with uv
Installation with uv (recommended package manager)
uv add "fastapi[standard]"
uv add uvicorn[standard]
uv add python-jose[cryptography]
uv add passlib[bcrypt]
uv add python-multipart
uv add sqlalchemy
uv add asyncpg
uv add python-dotenv
uv add pytest
uv add pytest-asyncio
uv add httpx
Alternative: Install all dependencies at once
uv add "fastapi[standard]" uvicorn python-jose passlib python-multipart sqlalchemy asyncpg python-dotenv pytest pytest-asyncio httpx
Recommended Project Structure
my-fastapi-project/
├── main.py # Application entry point
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI app instance
│ ├── api/
│ │ ├── __init__.py
│ │ ├── v1/
│ │ │ ├── __init__.py
│ │ │ └── routes/
│ │ │ ├── __init__.py
│ │ │ ├── users.py
│ │ │ └── items.py
│ ├── models/ # Pydantic models
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ ├── schemas/ # Database schemas
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ ├── database/
│ │ ├── __init__.py
│ │ └── database.py
│ ├── auth/
│ │ ├── __init__.py
│ │ └── auth.py
│ └── utils/
│ ├── __init__.py
│ └── helpers.py
├── tests/
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_users.py
│ └── test_items.py
├── requirements.txt
├── .env
├── .gitignore
└── README.md
Core FastAPI Application Structure
main.py - Application Entry Point with Advanced Configuration
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import uvicorn
import os
import logging
import datetime
from dotenv import load_dotenv
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("Starting up...")
yield
logger.info("Shutting down...")
app = FastAPI(
title="My FastAPI Application",
description="A comprehensive API built with FastAPI",
version="1.0.0",
lifespan=lifespan,
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json",
)
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=os.getenv("ALLOWED_ORIGINS", ).split(),
allow_credentials=,
allow_methods=[],
allow_headers=[],
allow_credentials=,
allow_headers=[],
)
():
logger.info()
response = call_next(request)
logger.info()
response
():
logger.error()
JSONResponse(
status_code=,
content={: }
)
():
{: , : }
():
{: , : datetime.datetime.utcnow()}
app.api.v1 router api_v1_router
app.include_router(api_v1_router, prefix=)
__name__ == :
uvicorn.run(
,
host=,
port=(os.getenv(, )),
reload=(os.getenv(, ).lower() == ),
log_level=os.getenv(, )
)
Advanced Routing and Path Operations
from fastapi import APIRouter, Path, Query, Body, status
from typing import List, Optional, Union
from pydantic import BaseModel, Field
import datetime
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/{user_id}", summary="Get user by ID")
async def get_user(
user_id: int = Path(..., ge=1, description="The ID of the user to retrieve"),
):
"""
Retrieve a user by ID.
- **user_id**: The unique identifier of the user
"""
return {"user_id": user_id}
@router.get("/", summary="Get multiple users")
async def get_users(
skip: int = Query(0, ge=0, description="Number of users to skip"),
limit: int = Query(100, ge=1, le=1000, description="Maximum number of users to return"),
q: [] = Query(),
active_only: = Query(),
):
{: skip, : limit, : q, : active_only}
():
user
():
{: user_id, : user_update, : notify}
():
fastapi UploadFile, File
typing
():
contents = file.read()
{
: file.filename,
: file.content_type,
: (contents)
}
():
results = []
file files:
contents = file.read()
results.append({
: file.filename,
: file.content_type,
: (contents)
})
{: results}
Dependency Injection Advanced Patterns
from fastapi import Depends, Header, HTTPException
from typing import Optional
import secrets
async def common_parameters(
q: Optional[str] = None,
skip: int = 0,
limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
async def verify_token(x_token: str = Header(...)):
if not secrets.compare_digest(x_token, "fake-super-secret-token"):
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: str = Header(...)):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
from sqlalchemy.ext.asyncio import AsyncSession
() -> AsyncSession:
AsyncSessionLocal() session:
:
session
:
session.close()
():
user = get_user_from_token(db, token)
user:
HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=,
headers={: },
)
user
():
current_user.is_active:
HTTPException(status_code=, detail=)
current_user
():
current_user
Background Tasks
from fastapi import BackgroundTasks
import asyncio
def send_email_task(email: str, message: str):
"""Simulate sending an email in the background"""
print(f"Sending email to {email}: {message}")
time.sleep(2)
print("Email sent!")
@router.post("/send-email")
async def send_email(
email: str,
background_tasks: BackgroundTasks
):
"""
Send email in background task.
"""
background_tasks.add_task(send_email_task, email, "Welcome to our service!")
return {"message": "Email will be sent in background"}
Custom Response Classes
from fastapi.responses import ORJSONResponse, UJSONResponse, HTMLResponse
from fastapi import Response
@router.get("/optimized-json", response_class=ORJSONResponse)
async def get_optimized_json():
return {"message": "This uses orjson for faster serialization"}
@router.get("/html", response_class=HTMLResponse)
async def get_html():
return """
<html>
<head>
<title>FastAPI HTML Response</title>
</head>
<body>
<h1>Hello from FastAPI!</h1>
</body>
</html>
"""
@router.get("/custom-response")
async def get_custom_response(response: Response):
response.headers["X-Custom-Header"] = "Custom value"
return {"message": "Response with custom header"}
Authentication Systems
JWT Authentication with Security Best Practices
from datetime import datetime, timedelta
from typing import Optional
from fastapi import Depends, HTTPException, status, Request
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
import secrets
import os
import bcrypt
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=True)
SECRET_KEY = os.getenv("SECRET_KEY", secrets.token_urlsafe(32))
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7
class Token(BaseModel):
access_token: str
token_type: str
refresh_token: Optional[str] = None
class TokenData(BaseModel):
username: Optional[str] = None
scopes: list[str] = []
class ():
username:
email: [] =
full_name: [] =
disabled: [] =
():
is_active: =
is_admin: =
():
hashed_password:
():
username:
email:
password:
full_name: [] =
():
pwd_context.verify(plain_password, hashed_password)
():
pwd_context.(password)
() -> [UserInDB]:
username db:
user_dict = db[username]
UserInDB(**user_dict)
():
user = get_user(db, username)
user:
verify_password(password, )
verify_password(password, user.hashed_password):
user
():
to_encode = data.copy()
expires_delta:
expire = datetime.utcnow() + expires_delta
:
expire = datetime.utcnow() + timedelta(minutes=)
to_encode.update({: expire, : })
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
encoded_jwt
():
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
to_encode.update({: expire, : })
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
encoded_jwt
():
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=,
headers={: },
)
:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: = payload.get()
token_type: = payload.get()
username :
credentials_exception
token_type != :
credentials_exception
token_data = TokenData(username=username)
JWTError:
credentials_exception
user = get_user(fake_users_db, username=token_data.username)
user :
credentials_exception
user
():
current_user.disabled:
HTTPException(status_code=, detail=)
current_user
():
current_user.is_admin:
HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=
)
current_user
fake_users_db = {
: {
: ,
: ,
: ,
: get_password_hash(),
: ,
: ,
: ,
},
: {
: ,
: ,
: ,
: get_password_hash(),
: ,
: ,
: ,
}
}
():
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
user:
HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=,
headers={: },
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={: user.username, : []},
expires_delta=access_token_expires
)
refresh_token = create_refresh_token(data={: user.username})
{
: access_token,
: ,
: refresh_token
}
():
current_user
():
:
payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
username: = payload.get()
token_type: = payload.get()
username token_type != :
HTTPException(status_code=, detail=)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
new_access_token = create_access_token(
data={: username, : []},
expires_delta=access_token_expires
)
{: new_access_token, : }
JWTError:
HTTPException(status_code=, detail=)
OAuth2 with Authorization Code Flow (Google, GitHub, etc.)
from fastapi import Request
from fastapi.responses import RedirectResponse, JSONResponse
from authlib.integrations.starlette_client import OAuth
import os
from urllib.parse import urlencode
oauth = OAuth()
oauth.register(
name='google',
client_id=os.getenv('GOOGLE_CLIENT_ID'),
client_secret=os.getenv('GOOGLE_CLIENT_SECRET'),
server_metadata_url='https://accounts.google.com/.well-known/openid_configuration',
client_kwargs={
'scope': 'openid email profile'
}
)
oauth.register(
name='github',
client_id=os.getenv('GITHUB_CLIENT_ID'),
client_secret=os.getenv('GITHUB_CLIENT_SECRET'),
access_token_url='https://github.com/login/oauth/token',
authorize_url='https://github.com/login/oauth/authorize',
api_base_url='https://api.github.com/',
client_kwargs={'scope': 'user:email'}
)
@app.route('/auth/{provider}/login')
async def auth_login(request: Request, provider: str):
if provider not in ['google', 'github']:
raise HTTPException(status_code=404, detail="Provider not supported")
redirect_uri = request.url_for()
oauth.create_client(provider).authorize_redirect(request, redirect_uri)
():
:
token = oauth.create_client(provider).authorize_access_token(request)
user_info = token.get() oauth.create_client(provider).userinfo(token=token)
access_token = create_access_token(data={: user_info.get(, user_info.get())})
params = urlencode({: access_token})
RedirectResponse(url=)
Exception e:
()
JSONResponse(status_code=, content={: })
base64
hashlib
():
provider [, ]:
HTTPException(status_code=, detail=)
code_verifier = secrets.token_urlsafe()
code_challenge = base64.urlsafe_b64encode(
hashlib.sha256(code_verifier.encode()).digest()
).decode().rstrip()
request.session[] = code_verifier
redirect_uri = request.url_for()
oauth.create_client(provider).authorize_redirect(
request,
redirect_uri,
code_challenge=code_challenge,
code_challenge_method=
)
API Key Authentication
from fastapi.security import APIKeyHeader, APIKeyQuery
from typing import Optional
from fastapi import Security
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
api_key_query = APIKeyQuery(name="api_key", auto_error=False)
API_KEYS = os.getenv("API_KEYS", "").split(",")
async def get_api_key(api_key_header: str = Security(api_key_header),
api_key_query: str = Security(api_key_query)):
api_key = api_key_header or api_key_query
if api_key and api_key in API_KEYS:
return api_key
else:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API Key"
)
@app.get("/api/protected-endpoint")
async def protected_endpoint(api_key: str = Security(get_api_key)):
return {"message": "Access granted with API key", "key": api_key[:8] + "..." if api_key else }
Better Auth Integration (Alternative approach)
from fastapi import Request, HTTPException
import jwt
import os
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
async def verify_better_auth_session(request: Request):
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Not authenticated")
token = auth_header.split(" ")[1]
try:
public_key_pem = os.getenv("BETTER_AUTH_PUBLIC_KEY")
if not public_key_pem:
raise HTTPException(status_code=500, detail="Public key not configured")
public_key = serialization.load_pem_public_key(
public_key_pem.encode(),
backend=default_backend()
)
payload = jwt.decode(
token,
public_key,
algorithms=[]
)
{
: payload.get(),
: payload.get(),
: payload.get(),
: payload.get()
}
jwt.InvalidTokenError:
HTTPException(status_code=, detail=)
Exception e:
()
HTTPException(status_code=, detail=)
():
{: , : user_info}
Role-Based Access Control (RBAC)
from enum import Enum
from functools import wraps
from typing import List
class Role(str, Enum):
USER = "user"
MODERATOR = "moderator"
ADMIN = "admin"
SUPER_ADMIN = "super_admin"
def require_role(required_roles: List[Role]):
def role_checker(current_user: User = Depends(get_current_active_user)):
if current_user.is_admin or current_user.username == "admin":
return current_user
user_role = getattr(current_user, 'role', Role.USER)
if user_role not in required_roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Access denied. Required roles: {', '.join([role.value for role in required_roles])}"
)
return current_user
return role_checker
@app.get("/admin-panel", dependencies=[Depends(require_role([Role.ADMIN, Role.SUPER_ADMIN]))])
async def admin_panel():
{: , : current_user.username}
():
{: , : current_user.username}
Security Headers and Rate Limiting
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/limited-endpoint")
@limiter.limit("5/minute")
async def limited_endpoint(request: Request):
return {"message": "This endpoint is rate limited"}
from starlette.middleware.base import BaseHTTPMiddleware
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
return response
app.add_middleware(SecurityHeadersMiddleware)
Pydantic Models
Advanced Pydantic Models with Validation
from pydantic import (
BaseModel,
EmailStr,
field_validator,
field_serializer,
model_validator,
ConfigDict,
Field,
HttpUrl,
AnyUrl
)
from typing import Optional, List, Dict, Any, Union
from datetime import datetime, date
from enum import Enum
import re
import uuid
class DatabaseConfig(BaseModel):
host: str = Field(..., description="Database host")
port: int = Field(5432, ge=1, le=65535, description="Database port")
database: str = Field(..., description="Database name")
username: str = Field(..., description="Database username")
password: str = Field(..., description="Database password")
model_config = ConfigDict(extra="forbid")
class PyObjectId(str):
@classmethod
def __get_pydantic_core_schema__(cls, source_type, handler):
return handler(str)
class UserRole(, Enum):
USER =
MODERATOR =
ADMIN =
SUPER_ADMIN =
():
email: EmailStr
username: = Field(..., min_length=, max_length=, pattern=)
full_name: [] = Field(, min_length=, max_length=)
bio: [] = Field(, max_length=)
avatar_url: [HttpUrl] =
():
re.(, v):
ValueError()
v
():
password: = Field(..., min_length=)
confirm_password:
():
.password != .confirm_password:
ValueError()
():
(v) < :
ValueError()
re.search(, v):
ValueError()
re.search(, v):
ValueError()
re.search(, v):
ValueError()
re.search(, v):
ValueError()
v
():
email: [EmailStr] =
username: [] = Field(, min_length=, max_length=)
full_name: [] = Field(, min_length=, max_length=)
bio: [] = Field(, max_length=)
avatar_url: [HttpUrl] =
():
: PyObjectId
created_at: datetime
updated_at: [datetime] =
is_active: =
role: UserRole = UserRole.USER
email_verified: =
model_config = ConfigDict(from_attributes=)
() -> :
(value)
() -> :
value.isoformat() value
():
user: UserResponse
posts_count: =
followers_count: =
following_count: =
is_following: =
(, Enum):
DRAFT =
PUBLISHED =
ARCHIVED =
():
title: = Field(..., min_length=, max_length=)
description: [] = Field(, max_length=)
price: = Field(..., gt=)
tags: [] = Field(default_factory=, max_items=)
metadata: [, ] = Field(default_factory=)
():
(v) > :
ValueError()
tag v:
(tag) > :
ValueError()
v
():
is_public: =
():
title: [] = Field(, min_length=, max_length=)
description: [] = Field(, max_length=)
price: [] = Field(, gt=)
tags: [[]] = Field(, max_items=)
status: [ItemStatus] =
is_public: [] =
metadata: [[, ]] = Field()
():
: PyObjectId
owner_id: PyObjectId
created_at: datetime
updated_at: [datetime] =
status: ItemStatus = ItemStatus.DRAFT
is_public: =
views_count: =
likes_count: =
is_liked: =
model_config = ConfigDict(from_attributes=)
() -> :
(value)
() -> :
value.isoformat() value
():
page: = Field(, ge=)
limit: = Field(, ge=, le=)
sort_by: =
sort_order: = Field(, pattern=)
() -> :
(.page - ) * .limit
():
total:
page:
limit:
pages:
has_next:
has_prev:
():
items: [ItemResponse]
pagination: PaginationResponse
():
filename:
content_type:
size: = Field(..., gt=, le= * * )
url: HttpUrl
():
success_count:
failure_count:
errors: [[, ]] = Field(default_factory=)
():
success: =
message:
data: [] =
error_code: [] =
():
access_token:
token_type:
refresh_token: [] =
expires_in: [] =
():
username: [] =
scopes: [] = Field(default_factory=)
():
detail:
error_code: [] =
timestamp: datetime = Field(default_factory=datetime.utcnow)
():
app_name: =
app_version: =
debug: =
database_url:
secret_key:
algorithm: =
access_token_expire_minutes: =
refresh_token_expire_days: =
allowed_origins: [] = Field(default_factory=)
model_config = ConfigDict(
env_file=,
env_file_encoding=,
extra=
)
():
event_type:
event_id: = Field(default_factory=: (uuid.uuid4()))
timestamp: datetime = Field(default_factory=datetime.utcnow)
data: [, ]
signature: [] =
():
q: = Field(..., min_length=, max_length=, description=)
filters: [, ] = Field(default_factory=)
page: = Field(, ge=)
limit: = Field(, ge=, le=)
sort_by: [] =
sort_order: = Field(, pattern=)
Custom Pydantic Data Types and Validators
from pydantic import BaseModel, field_validator
from typing import Optional
import phonenumbers
from phonenumbers import NumberParseException
import ipaddress
from pydantic.functional_validators import AfterValidator
from typing_extensions import Annotated
import json
def validate_phone_number(v: str) -> str:
try:
parsed = phonenumbers.parse(v, None)
if not phonenumbers.is_valid_number(parsed):
raise ValueError("Invalid phone number")
return phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164)
except NumberParseException:
raise ValueError("Invalid phone number format")
PhoneNumber = Annotated[str, AfterValidator(validate_phone_number)]
def validate_ip_address(v: str) -> str:
try:
ipaddress.IPv4Address(v)
return v
except ipaddress.AddressValueError:
raise ValueError("Invalid IP address")
IPAddress = Annotated[str, AfterValidator(validate_ip_address)]
() -> :
(v, ):
v
:
json.loads(v)
json.JSONDecodeError:
ValueError()
JSONType = Annotated[[, ], AfterValidator(validate_json)]
():
phone: [PhoneNumber] =
ip_address: [IPAddress] =
metadata: [JSONType] =
tags: [] = Field(default_factory=)
():
name:
price:
category:
tags: [] = []
inventory: = Field(..., ge=)
is_available: =
():
.inventory == :
.is_available =
.inventory > .is_available:
()
():
(v) != ((v)):
ValueError()
v
Nested Models and Relationships
from typing import List, Optional
from pydantic import BaseModel, Field
class Address(BaseModel):
street: str
city: str
state: str
zip_code: str = Field(..., pattern=r'^\d{5}(-\d{4})?$')
country: str = "US"
class Company(BaseModel):
id: Optional[PyObjectId] = None
name: str
description: Optional[str] = None
website: Optional[HttpUrl] = None
address: Optional[Address] = None
class UserWithCompany(UserResponse):
company: Optional[Company] = None
colleagues: List[UserResponse] = Field(default_factory=list)
class OrderItem(BaseModel):
product_id: PyObjectId
quantity: int = Field(..., ge=1)
price: float = Field(..., ge=0)
class Order(BaseModel):
id: Optional[PyObjectId] =
user_id: PyObjectId
items: [OrderItem]
total_amount: = Field(..., ge=)
status: =
shipping_address: Address
billing_address: [Address] =
created_at: datetime = Field(default_factory=datetime.utcnow)
():
v:
ValueError()
v
() -> :
(item.price * item.quantity item .items)
Database Integration
SQLAlchemy Async Setup with Advanced Configuration
from sqlalchemy import (
create_engine,
Column,
Integer,
String,
DateTime,
Boolean,
Float,
Text,
Index,
ForeignKey,
UniqueConstraint,
text
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, backref
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.pool import AsyncAdaptedQueuePool
from datetime import datetime
import os
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://user:password@localhost/dbname")
async_engine = create_async_engine(
DATABASE_URL,
poolclass=AsyncAdaptedQueuePool,
pool_size=20,
max_overflow=30,
pool_pre_ping=True,
pool_recycle=300,
echo=bool(os.getenv("DB_ECHO", "False").lower() == "true")
)
AsyncSessionLocal = sessionmaker(
async_engine,
class_=AsyncSession,
expire_on_commit=False
)
sync_engine = create_engine(
DATABASE_URL.replace("postgresql+asyncpg", "postgresql"),
pool_size=20,
max_overflow=30,
pool_pre_ping=True,
pool_recycle=,
echo=(os.getenv(, ).lower() == )
)
SessionLocal = sessionmaker(autocommit=, autoflush=, bind=sync_engine)
Base = declarative_base()
():
AsyncSessionLocal() db:
:
db
:
db.close()
():
db = SessionLocal()
:
db
:
db.close()
:
created_at = Column(DateTime, default=datetime.utcnow, nullable=)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=)
(Base, TimestampMixin):
__abstract__ =
= Column(Integer, primary_key=, index=)
():
__tablename__ =
email = Column(String(), unique=, index=, nullable=)
username = Column(String(), unique=, index=, nullable=)
full_name = Column(String())
hashed_password = Column(String(), nullable=)
is_active = Column(Boolean, default=, nullable=)
is_verified = Column(Boolean, default=, nullable=)
role = Column(String(), default=, nullable=)
items = relationship(, back_populates=, cascade=)
profile = relationship(, back_populates=, uselist=, cascade=)
__table_args__ = (
Index(, ),
Index(, ),
Index(, ),
)
():
__tablename__ =
user_id = Column(Integer, ForeignKey(), unique=, nullable=)
bio = Column(Text)
avatar_url = Column(String())
phone = Column(String())
birth_date = Column(DateTime)
user = relationship(, back_populates=)
():
__tablename__ =
title = Column(String(), nullable=)
description = Column(Text)
price = Column(Float, nullable=)
is_public = Column(Boolean, default=, nullable=)
status = Column(String(), default=, nullable=)
owner_id = Column(Integer, ForeignKey(), nullable=)
owner = relationship(, back_populates=)
tags = relationship(, back_populates=, cascade=)
__table_args__ = (
Index(, ),
Index(, ),
Index(, ),
)
():
__tablename__ =
item_id = Column(Integer, ForeignKey(), nullable=)
tag_name = Column(String(), nullable=)
item = relationship(, back_populates=)
__table_args__ = (
UniqueConstraint(, , name=),
Index(, ),
)
():
__tablename__ =
table_name = Column(String(), nullable=)
record_id = Column(Integer, nullable=)
action = Column(String(), nullable=)
old_values = Column(Text)
new_values = Column(Text)
user_id = Column(Integer, ForeignKey())
timestamp = Column(DateTime, default=datetime.utcnow, nullable=)
user = relationship()
():
async_engine.begin() conn:
conn.run_sync(Base.metadata.create_all)
logger.info()
():
async_engine.begin() conn:
conn.run_sync(Base.metadata.drop_all)
logger.info()
contextlib asynccontextmanager
():
AsyncSessionLocal() session:
:
session
session.commit()
Exception:
session.rollback()
:
session.close()
Async Database Operations and Repository Pattern
from typing import List, Optional, Dict, Any
from sqlalchemy import select, update, delete, and_, or_, func
from sqlalchemy.orm import selectinload
from sqlalchemy.exc import IntegrityError
import json
class BaseRepository:
def __init__(self, session: AsyncSession):
self.session = session
async def create(self, model, **kwargs):
"""Create a new record"""
try:
instance = model(**kwargs)
self.session.add(instance)
await self.session.commit()
await self.session.refresh(instance)
return instance
except IntegrityError as e:
await self.session.rollback()
raise ValueError(f"Integrity error: {str(e)}")
async def get_by_id(self, model, id: int):
"""Get a record by ID"""
stmt = select(model).where(model.id == )
result = .session.execute(stmt)
result.scalar_one_or_none()
():
stmt = select(model).where(model..in_(ids))
result = .session.execute(stmt)
result.scalars().()
():
stmt = update(model).where(model. == ).values(**kwargs)
result = .session.execute(stmt)
result.rowcount == :
.session.commit()
.get_by_id(model, )
():
stmt = delete(model).where(model. == )
result = .session.execute(stmt)
result.rowcount == :
.session.commit()
():
stmt = select(model)
filters:
conditions = []
key, value filters.items():
(model, key):
(value, ):
conditions.append((model, key).in_(value))
:
conditions.append((model, key) == value)
conditions:
stmt = stmt.where(and_(*conditions))
stmt = stmt.offset(skip).limit(limit)
result = .session.execute(stmt)
result.scalars().()
():
stmt = select(func.count(model.))
filters:
conditions = []
key, value filters.items():
(model, key):
(value, ):
conditions.append((model, key).in_(value))
:
conditions.append((model, key) == value)
conditions:
stmt = stmt.where(and_(*conditions))
result = .session.execute(stmt)
result.scalar_one()
():
():
stmt = select(User).where(User.username == username)
result = .session.execute(stmt)
result.scalar_one_or_none()
():
stmt = select(User).where(User.email == email)
result = .session.execute(stmt)
result.scalar_one_or_none()
():
stmt = select(User).where(User.is_active == ).offset(skip).limit(limit)
result = .session.execute(stmt)
result.scalars().()
():
search_filter = or_(
User.username.contains(query),
User.email.contains(query),
User.full_name.contains(query) query
)
stmt = select(User).where(search_filter).offset(skip).limit(limit)
result = .session.execute(stmt)
result.scalars().()
():
():
stmt = select(Item).where(Item.owner_id == owner_id).offset(skip).limit(limit)
result = .session.execute(stmt)
result.scalars().()
():
stmt = select(Item).where(Item.is_public == ).offset(skip).limit(limit)
result = .session.execute(stmt)
result.scalars().()
():
stmt = (
select(Item)
.join(ItemTag)
.where(ItemTag.tag_name.in_(tag_names))
.offset(skip)
.limit(limit)
)
result = .session.execute(stmt)
result.scalars().()
():
stmt = (
select(Item)
.options(selectinload(Item.owner))
.offset(skip)
.limit(limit)
)
result = .session.execute(stmt)
result.scalars().()
fastapi Depends, HTTPException, status
() -> UserRepository:
UserRepository(db)
() -> ItemRepository:
ItemRepository(db)
():
user = repo.get_by_id(User, user_id)
user:
HTTPException(status_code=, detail=)
user
():
items = repo.get_by_owner(user_id, skip, limit)
items
Database Migrations with Alembic
"""
[alembic]
# path to migration scripts
script_location = alembic
# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s
# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .
# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python-dateutil library that can be
# installed by adding `alembic[tz]` to the pip requirements
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =
# max length of characters to apply to the
# "slug" field
# max_length = 40
# version number format
# version_num_format = %04d
# version path separator; As mentioned above, this is the character used to split
# version_locations. The default within new alembic.ini files is "os", which uses
# os.pathsep. If this key is omitted entirely, it falls back to the legacy
# behavior of splitting on spaces and/or commas.
# Valid values for version_path_separator are:
#
# version_path_separator = :
# version_path_separator = ;
# version_path_separator = space
version_path_separator = os # Use os.pathsep. Default configuration used for new projects.
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
sqlalchemy.url = driver://user:pass@localhost/dbname
[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are invoked
# automatically whenever a new revision file is created.
# Options include:
#
# hooks = black, isort
# black.type = exec
# black.executable = black
# black.args = -l 79 REVISION_SCRIPT_FILENAME
# isort.type = exec
# isort.executable = isort
# isort.args = REVISION_SCRIPT_FILENAME
#
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
"""
"""
Revision ID: abc123def456
Revises:
Create Date: 2023-10-01 12:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
revision = 'abc123def456'
down_revision = None
branch_labels = None
depends_on =
() -> :
op.create_table(
,
sa.Column(, sa.Integer(), nullable=),
sa.Column(, sa.String(length=), nullable=),
sa.Column(, sa.String(length=), nullable=),
sa.Column(, sa.String(length=), nullable=),
sa.Column(, sa.String(length=), nullable=),
sa.Column(, sa.Boolean(), nullable=, default=),
sa.Column(, sa.Boolean(), nullable=, default=),
sa.Column(, sa.String(length=), nullable=, default=),
sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
sa.PrimaryKeyConstraint(),
sa.UniqueConstraint(),
sa.UniqueConstraint()
)
op.create_table(
,
sa.Column(, sa.Integer(), nullable=),
sa.Column(, sa.Integer(), nullable=),
sa.Column(, sa.Text(), nullable=),
sa.Column(, sa.String(length=), nullable=),
sa.Column(, sa.String(length=), nullable=),
sa.Column(, sa.DateTime(), nullable=),
sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
sa.ForeignKeyConstraint([], [], ),
sa.PrimaryKeyConstraint(),
sa.UniqueConstraint()
)
op.create_table(
,
sa.Column(, sa.Integer(), nullable=),
sa.Column(, sa.String(length=), nullable=),
sa.Column(, sa.Text(), nullable=),
sa.Column(, sa.Float(), nullable=),
sa.Column(, sa.Boolean(), nullable=, default=),
sa.Column(, sa.String(length=), nullable=, default=),
sa.Column(, sa.Integer(), nullable=),
sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
sa.ForeignKeyConstraint([], [], ),
sa.PrimaryKeyConstraint()
)
op.create_index(, , [])
op.create_index(, , [])
op.create_index(, , [])
op.create_index(, , [])
op.create_index(, , [])
op.create_index(, , [])
() -> :
op.drop_index(, table_name=)
op.drop_index(, table_name=)
op.drop_index(, table_name=)
op.drop_index(, table_name=)
op.drop_index(, table_name=)
op.drop_index(, table_name=)
op.drop_table()
op.drop_table()
op.drop_table()
Database Connection Pooling and Performance Optimization
from sqlalchemy import event
from sqlalchemy.pool import Pool
import time
DATABASE_CONFIG = {
"pool_size": 20,
"max_overflow": 30,
"pool_pre_ping": True,
"pool_recycle": 300,
"pool_timeout": 30,
"echo": bool(os.getenv("DB_ECHO", "False").lower() == "true")
}
@event.listens_for(sync_engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
"""Set SQLite pragmas for performance (if using SQLite)"""
if "sqlite" in DATABASE_URL:
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.execute("PRAGMA journal_mode=WAL")
cursor.execute("PRAGMA synchronous=NORMAL")
cursor.close()
@event.listens_for(async_engine, "connect")
def set_async_sqlite_pragma(dbapi_connection, connection_record):
"""Set SQLite pragmas for async engine"""
if DATABASE_URL:
cursor = dbapi_connection.cursor()
cursor.execute()
cursor.execute()
cursor.execute()
cursor.close()
():
connection_record.start_time = time.time()
():
(connection_record, ):
total_time = time.time() - connection_record.start_time
total_time > :
logger.warning()
():
() -> [, ]:
sql =
result = .session.execute(text(sql), {: user_id})
row = result.fetchone()
row:
{
: row[],
: row[],
: row[],
: row[],
: (row[]) row[] ,
: row[]
}
() -> :
stmt = Item.__table__.insert()
result = .session.execute(stmt, items_data)
.session.commit()
result.rowcount
() -> [[, ]]:
conditions = []
params = {}
filters.get():
conditions.append()
params[] = filters[]
filters.get():
conditions.append()
params[] = filters[]
filters.get():
conditions.append()
params[] = filters[]
filters.get():
conditions.append()
params[] = filters[]
where_clause = conditions
sql =
params[] = filters.get(, )
params[] = filters.get(, )
result = .session.execute(text(sql), params)
rows = result.fetchall()
[
{
: row[],
: row[],
: row[],
: (row[]),
: row[],
: row[],
: row[],
: row[],
: row[]
}
row rows
]
Deployment and Testing Patterns
Production Deployment Configuration
import multiprocessing
bind = "0.0.0.0:8000"
backlog = 2048
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
timeout = 30
keepalive = 2
max_requests = 1000
max_requests_jitter = 100
preload_app = True
reload = False
accesslog = "-"
errorlog = "-"
loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'
proc_name = "fastapi_app"
limit_request_line = 4094
limit_request_fields = 100
limit_request_field_size = 8190
Docker Configuration for Production
# Dockerfile
FROM python:3.11-slim
# Install system dependencies
RUN apt-get update && apt-get install -y \
gcc \
g++ \
&& rm -rf /var/lib/apt/lists/*
# Install uv
RUN pip install uv
# Set working directory
WORKDIR /app
# Copy requirements and install dependencies
COPY requirements.txt .
RUN uv pip install -r requirements.txt
# Copy application code
COPY . .
# Create non-root user
RUN useradd --create-home --shell /bin/bash app \
&& chown -R app:app /app
USER app
# Expose port
EXPOSE 8000
# Health check
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# Run the application
CMD ["gunicorn", "main:app", "-c", "gunicorn.conf.py"]
Docker Compose for Development and Production
version: '3.8'
services:
app:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql+asyncpg://user:password@db:5432/myapp
- REDIS_URL=redis://redis:6379/0
- SECRET_KEY=your-super-secret-key-here
- DEBUG=False
depends_on:
- db
- redis
volumes:
- ./logs:/app/logs
restart: unless-stopped
db:
image: postgres:15
environment:
- POSTGRES_DB=myapp
- POSTGRES_USER=user
- POSTGRES_PASSWORD=password
volumes:
- postgres_data:/var/lib/postgresql/data
- ./init.sql:/docker-entrypoint-initdb.d/init.sql
ports:
- "5432:5432"
restart: unless-stopped
redis:
image:
Environment Configuration
import os
from pydantic import BaseModel, Field
from typing import Optional, List
class Settings(BaseModel):
app_name: str = "My FastAPI App"
app_version: str = "1.0.0"
debug: bool = Field(default=False, env="DEBUG")
environment: str = Field(default="development", env="ENVIRONMENT")
database_url: str = Field(..., env="DATABASE_URL")
database_pool_size: int = Field(default=20, env="DATABASE_POOL_SIZE")
database_max_overflow: int = Field(default=30, env="DATABASE_MAX_OVERFLOW")
secret_key: str = Field(..., env="SECRET_KEY")
algorithm: str = Field(default="HS256", env="ALGORITHM")
access_token_expire_minutes: int = Field(default=30, env="ACCESS_TOKEN_EXPIRE_MINUTES")
refresh_token_expire_days: int = Field(default=7, env="REFRESH_TOKEN_EXPIRE_DAYS")
allowed_origins: List[str] = Field(default=[], env=)
redis_url: [] = Field(default=, env=)
log_level: = Field(default=, env=)
log_format: = Field(default=, env=)
sentry_dsn: [] = Field(default=, env=)
mailgun_api_key: [] = Field(default=, env=)
:
env_file =
env_file_encoding =
case_sensitive =
settings = Settings()
contextlib asynccontextmanager
fastapi FastAPI
logging
logging.basicConfig(
level=(logging, settings.log_level.upper()),
=settings.log_format
)
():
()
init_db()
settings.redis_url:
()
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
debug=settings.debug,
lifespan=lifespan
)
Testing Framework Setup
import pytest
import asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.main import app
from app.database import Base
import os
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
@pytest.fixture(scope="session")
def event_loop():
"""Create an instance of the default event loop for each test case."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="session", autouse=True)
async def create_test_database():
"""Create test database before tests and clean up after."""
engine = create_async_engine(
TEST_DATABASE_URL,
poolclass=StaticPool,
connect_args={"check_same_thread": False}
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
engine.dispose()
():
AsyncClient(app=app, base_url=) ac:
ac
():
engine = create_async_engine(
TEST_DATABASE_URL,
poolclass=StaticPool,
connect_args={: }
)
async_session = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=
)
async_session() session:
session
session.rollback()
pytest
httpx AsyncClient
():
response = async_client.get()
response.status_code ==
response.json()
():
response = async_client.get()
response.status_code ==
response.json()[] ==
app.auth create_access_token
jwt
():
data = {: }
token = create_access_token(data=data)
payload = jwt.decode(token, os.getenv(), algorithms=[])
payload[] ==
():
token = create_access_token(data={: })
response = async_client.get(
,
headers={: }
)
response.status_code ==
app.models User
app.database get_async_db
sqlalchemy select
():
user = User(
email=,
username=,
hashed_password=
)
db_session.add(user)
db_session.commit()
db_session.refresh(user)
result = db_session.execute(select(User).where(User. == user.))
retrieved_user = result.scalar_one_or_none()
retrieved_user
retrieved_user.email ==
retrieved_user.username ==
app.models Item
app.schemas ItemCreate
():
user = User(
email=,
username=,
hashed_password=
)
db_session.add(user)
db_session.commit()
db_session.refresh(user)
item_data = {
: ,
: ,
: ,
: user.
}
response = async_client.post(, json=item_data)
response.status_code ==
item_response = response.json()
item_response[] ==
item_response[] ==
item_response[] == user.
():
user_data = {
: ,
: ,
: ,
:
}
response = async_client.post(, json=user_data)
response.status_code ==
login_data = {
: ,
:
}
response = async_client.post(, data=login_data)
response.status_code ==
token_data = response.json()
token_data
token_data[] ==
Performance Testing
from locust import HttpUser, task, between
import json
class FastAPIUser(HttpUser):
wait_time = between(1, 3)
def on_start(self):
"""Login before starting tasks."""
self.login()
def login(self):
"""Login to get JWT token."""
response = self.client.post("/token", data={
"username": "testuser",
"password": "testpassword"
})
if response.status_code == 200:
token_data = response.json()
self.token = token_data["access_token"]
self.headers = {"Authorization": f"Bearer {self.token}"}
else:
self.token = None
self.headers = {}
@task(3)
def get_items(self):
"""Get items endpoint."""
self.client.get("/items/", headers=self.headers)
():
item_data = {
: ,
: ,
:
}
.client.post(, json=item_data, headers=.headers)
():
.client.get(, headers=.headers)
():
():
user_data = {
: ,
: ,
:
}
result = benchmark(create_user)
benchmark.stats[] <
pytest
httpx AsyncClient
():
i ():
response = async_client.get()
response.status_code ==
():
malicious_payload = {
: ,
:
}
response = async_client.get(, params=malicious_payload)
response.status_code !=
():
response = async_client.get()
response.status_code ==
():
invalid_data = {
: ,
: -
}
response = async_client.post(, json=invalid_data)
response.status_code ==
