Website Backend Development with Python (FastAPI)
FastAPI is a modern Python framework that builds APIs around types. You declare a function with type hints, and FastAPI automatically generates validation through Pydantic, OpenAPI documentation, and JSON Schema. No manual documentation, no separate validators — everything is inferred from types.
FastAPI's performance on asynchronous I/O operations is comparable to Node.js. For CPU-bound tasks — that's a separate question; you need process pools or offloading to Celery.
Basics and Endpoint Structure
from fastapi import FastAPI, Depends, HTTPException, Query, Path, status
from pydantic import BaseModel, EmailStr, Field
from typing import Optional, List
import uvicorn
app = FastAPI(
title="My API",
version="1.0.0",
docs_url="/api/docs",
redoc_url="/api/redoc"
)
class ProductCreate(BaseModel):
name: str = Field(..., min_length=2, max_length=255)
price: float = Field(..., gt=0)
category_id: int
description: Optional[str] = None
class ProductResponse(BaseModel):
id: int
name: str
price: float
category_id: int
class Config:
from_attributes = True # allows creating from ORM objects
@app.get('/api/v1/products', response_model=List[ProductResponse])
async def list_products(
page: int = Query(1, ge=1),
limit: int = Query(20, ge=1, le=100),
category_id: Optional[int] = Query(None),
db: AsyncSession = Depends(get_db)
):
offset = (page - 1) * limit
query = select(Product).offset(offset).limit(limit)
if category_id:
query = query.where(Product.category_id == category_id)
result = await db.execute(query)
return result.scalars().all()
@app.post('/api/v1/products', response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
async def create_product(
body: ProductCreate,
current_user: User = Depends(require_role('admin')),
db: AsyncSession = Depends(get_db)
):
product = Product(**body.model_dump())
db.add(product)
await db.commit()
await db.refresh(product)
return product
Dependency Injection
DI in FastAPI works through Depends — one of the best mechanisms in the Python ecosystem:
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
# Database connection per request
async_engine = create_async_engine(settings.DATABASE_URL, pool_size=10)
async def get_db():
async with AsyncSession(async_engine) as session:
try:
yield session
except Exception:
await session.rollback()
raise
finally:
await session.close()
# Authentication
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/api/auth/token')
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> User:
try:
payload = jwt.decode(token, settings.JWT_SECRET, algorithms=['HS256'])
user_id: int = payload.get('sub')
except JWTError:
raise HTTPException(status_code=401, detail='Invalid token')
user = await db.get(User, user_id)
if not user or not user.is_active:
raise HTTPException(status_code=401, detail='Inactive user')
return user
def require_role(*roles: str):
async def checker(user: User = Depends(get_current_user)) -> User:
if user.role not in roles:
raise HTTPException(status_code=403, detail='Insufficient permissions')
return user
return checker
SQLAlchemy 2.0 (async)
FastAPI works well with SQLAlchemy 2.0 in async mode:
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, Numeric, ForeignKey, DateTime, func
class Base(DeclarativeBase):
pass
class Product(Base):
__tablename__ = 'products'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(255))
slug: Mapped[str] = mapped_column(String(255), unique=True)
price: Mapped[float] = mapped_column(Numeric(10, 2))
category_id: Mapped[int | None] = mapped_column(ForeignKey('categories.id'), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
category: Mapped['Category'] = relationship(back_populates='products', lazy='selectin')
lazy='selectin' for relationships — the best choice in async mode, avoids N+1 without explicit joins.
Background Tasks
Simple tasks — through BackgroundTasks, heavy ones — through Celery:
from fastapi import BackgroundTasks
import asyncio
# Light async tasks right in the request
@app.post('/api/orders/{order_id}/confirm')
async def confirm_order(
order_id: int,
background_tasks: BackgroundTasks,
db: AsyncSession = Depends(get_db)
):
order = await get_order_or_404(order_id, db)
order.status = 'confirmed'
await db.commit()
# Don't wait for completion
background_tasks.add_task(send_confirmation_email, order.user.email, order_id)
background_tasks.add_task(update_inventory, order.items)
return {'status': 'confirmed'}
# Heavy tasks — Celery
from celery import Celery
celery = Celery(__name__, broker=settings.REDIS_URL)
@celery.task(name='generate_report')
def generate_report(user_id: int, date_range: dict):
# CPU-intensive processing
pass
Middleware and CORS
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=['*'],
allow_headers=['*']
)
@app.middleware('http')
async def add_process_time(request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start
response.headers['X-Process-Time'] = str(round(duration * 1000, 2))
return response
WebSocket
from fastapi import WebSocket, WebSocketDisconnect
class ConnectionManager:
def __init__(self):
self.active: dict[int, list[WebSocket]] = {}
async def connect(self, user_id: int, ws: WebSocket):
await ws.accept()
self.active.setdefault(user_id, []).append(ws)
async def broadcast_to_user(self, user_id: int, message: dict):
for ws in self.active.get(user_id, []):
await ws.send_json(message)
manager = ConnectionManager()
@app.websocket('/ws/notifications')
async def notifications_ws(
websocket: WebSocket,
token: str = Query(...),
):
user = await verify_ws_token(token)
await manager.connect(user.id, websocket)
try:
while True:
await websocket.receive_text() # keep connection
except WebSocketDisconnect:
manager.disconnect(user.id, websocket)
Deployment
FastAPI runs through Uvicorn or Gunicorn with Uvicorn workers:
# Production: multiple workers
gunicorn app.main:app \
-w 4 \
-k uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--timeout 30 \
--keepalive 5
Development Timelines
- Framework + models + auth — 4–7 days
- API endpoints + validation — 1–2 weeks
- Integrations and background tasks — 1–2 weeks
- Tests (pytest + httpx AsyncClient) — 5–7 days
API for a mid-scale website: 4–8 weeks. FastAPI wins when the team knows Python and needs auto-documentation — Swagger/ReDoc is generated from code effortlessly.