Python

FastAPI Complete Guide

Master FastAPI development from setup to production including async programming, REST APIs, authentication, database integration, testing, performance optimisation, and enterprise deployment strategies.

What you'll learn

Build high-performance async REST APIs with FastAPI
Validate request/response with Pydantic v2 models
Implement OAuth2 + JWT authentication
Integrate SQLAlchemy 2.0 async ORM with PostgreSQL
Write tests with pytest-asyncio and httpx
Add background tasks and WebSocket support
Deploy with Docker and uvicorn workers

Setup & First API

Python

1pip install fastapi uvicorn[standard] pydantic-settings
2
3# main.py
4from fastapi import FastAPI
5from contextlib import asynccontextmanager
6
7@asynccontextmanager
8async def lifespan(app: FastAPI):
9    # startup
10    yield
11    # shutdown
12
13app = FastAPI(title="My API", version="1.0.0", lifespan=lifespan)
14
15@app.get("/health")
16async def health() -> dict[str, str]:
17    return {"status": "ok"}
18
19# Run
20uvicorn main:app --reload --port 8000

Pydantic Models

Python

1from pydantic import BaseModel, EmailStr, field_validator
2from datetime import datetime
3
4class UserCreate(BaseModel):
5    name:     str
6    email:    EmailStr
7    password: str
8
9    @field_validator('password')
10    @classmethod
11    def password_strength(cls, v: str) -> str:
12        if len(v) < 8:
13            raise ValueError('Password must be at least 8 characters')
14        return v
15
16class UserResponse(BaseModel):
17    id:         int
18    name:       str
19    email:      EmailStr
20    created_at: datetime
21
22    model_config = {"from_attributes": True}
23
24@app.post("/users", response_model=UserResponse, status_code=201)
25async def create_user(data: UserCreate) -> UserResponse:
26    ...

Dependency Injection

Python

1from fastapi import Depends
2from sqlalchemy.ext.asyncio import AsyncSession
3from .database import async_session_maker
4
5async def get_db() -> AsyncSession:
6    async with async_session_maker() as session:
7        yield session
8
9# Inject into any route
10@app.get("/courses")
11async def list_courses(db: AsyncSession = Depends(get_db)):
12    result = await db.execute(select(Course).where(Course.is_published == True))
13    return result.scalars().all()

JWT Authentication

Python

1pip install python-jose[cryptography] passlib[bcrypt]
2
3from jose import JWTError, jwt
4from passlib.context import CryptContext
5from fastapi.security import OAuth2PasswordBearer
6
7pwd_context   = CryptContext(schemes=["bcrypt"])
8oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
9SECRET_KEY    = os.environ["JWT_SECRET"]
10ALGORITHM     = "HS256"
11
12def create_access_token(sub: str) -> str:
13    expire = datetime.utcnow() + timedelta(minutes=15)
14    return jwt.encode({"sub": sub, "exp": expire}, SECRET_KEY, algorithm=ALGORITHM)
15
16async def get_current_user(token: str = Depends(oauth2_scheme), db=Depends(get_db)):
17    try:
18        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
19        user_id = payload.get("sub")
20    except JWTError:
21        raise HTTPException(status_code=401, detail="Invalid token")
22    user = await db.get(User, user_id)
23    if not user:
24        raise HTTPException(status_code=401)
25    return user

Async Database

Python

1pip install sqlalchemy asyncpg alembic
2
3# database.py
4from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
5from sqlalchemy.orm import DeclarativeBase
6
7DATABASE_URL = os.environ["DATABASE_URL"].replace("postgresql://", "postgresql+asyncpg://")
8engine = create_async_engine(DATABASE_URL, pool_size=10, max_overflow=20)
9async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
10
11class Base(DeclarativeBase): pass
12
13# models.py
14from sqlalchemy.orm import Mapped, mapped_column
15
16class Course(Base):
17    __tablename__ = "courses"
18    id:           Mapped[int]  = mapped_column(primary_key=True)
19    title:        Mapped[str]  = mapped_column(index=True)
20    is_published: Mapped[bool] = mapped_column(default=False)

Testing

Python

1pip install pytest pytest-asyncio httpx
2
3# tests/test_courses.py
4import pytest
5from httpx import AsyncClient, ASGITransport
6from main import app
7
8@pytest.mark.asyncio
9async def test_list_courses():
10    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
11        response = await client.get("/courses")
12    assert response.status_code == 200
13    assert isinstance(response.json(), list)

Deployment

DOCKERFILE

1# Dockerfile
2FROM python:3.12-slim
3WORKDIR /app
4COPY requirements.txt .
5RUN pip install --no-cache-dir -r requirements.txt
6COPY . .
7RUN adduser --disabled-password appuser && chown -R appuser /app
8USER appuser
9EXPOSE 8000
10CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]