FastAPI Complete Guide | Python REST API, Async, Pydantic & Auto Docs

· 30분 읽기 · 수정 intermediate tutorial
FastAPI Complete Guide | Python REST API, Async, Pydantic & Auto Docs

이 글의 핵심

FastAPI is a modern, high-performance Python web framework for building APIs. It matches Node.js and Go in speed, with automatic OpenAPI docs and Pydantic type safety built in.

Why FastAPI?

FastAPI is the fastest-growing Python web framework for building APIs. If you’re migrating from Flask, you’ll gain:

  • 3× faster throughput (async I/O, ASGI)
  • Automatic API docs (Swagger UI + ReDoc, zero config)
  • Type safety via Pydantic — catch bugs at startup, not runtime
  • Built-in validation — no more manual request parsing
Flask:   1,000 req/sec (sync, WSGI)
Django:    800 req/sec (sync, WSGI)
FastAPI: 3,000 req/sec (async, ASGI)

Installation

pip install fastapi uvicorn[standard]

Your First API (30 seconds)

The root function is implemented below. It handles the core logic described above:

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello, FastAPI!"}

@app.get("/items/{item_id}")
async def get_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}
uvicorn main:app --reload
# → http://localhost:8000
# → http://localhost:8000/docs  (Swagger UI)
# → http://localhost:8000/redoc (ReDoc)

Pydantic Models — Request & Response Validation

Pydantic is FastAPI’s core for data validation and serialization.

The create_user function is implemented below. It handles the core logic described above:

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime

app = FastAPI()

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)
    bio: str | None = None

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime

    class Config:
        from_attributes = True  # ORM mode

@app.post("/users/", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    # Validation is automatic — invalid data returns 422
    # Save to DB here...
    return UserResponse(
        id=1,
        name=user.name,
        email=user.email,
        created_at=datetime.now()
    )

What Pydantic does automatically:

  • Rejects invalid data with clear error messages (422 response)
  • Converts types (e.g., "42" → 42 for int fields)
  • Documents the schema in Swagger UI

Routing & HTTP Methods

The list_items function is implemented below. It handles the core logic described above:

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI()

# In-memory store for this example
items: dict[int, dict] = {}
next_id = 1

class Item(BaseModel):
    name: str
    price: float
    in_stock: bool = True

class ItemUpdate(BaseModel):
    name: str | None = None
    price: float | None = None
    in_stock: bool | None = None

@app.get("/items/", response_model=list[dict])
async def list_items(skip: int = 0, limit: int = 10):
    return list(items.values())[skip : skip + limit]

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    global next_id
    item_dict = {"id": next_id, **item.dict()}
    items[next_id] = item_dict
    next_id += 1
    return item_dict

@app.patch("/items/{item_id}")
async def update_item(item_id: int, item: ItemUpdate):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    update_data = item.dict(exclude_unset=True)
    items[item_id].update(update_data)
    return items[item_id]

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    del items[item_id]

Async/Await — When to Use It

The get_weather function is implemented below. It handles the core logic described above:

import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()

# ✅ Use async for I/O-bound operations
@app.get("/weather/{city}")
async def get_weather(city: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.openweathermap.org/data/2.5/weather",
            params={"q": city, "appid": "your-key"}
        )
        return response.json()

# ✅ Async DB query (using asyncpg, SQLAlchemy async, etc.)
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # await db.fetch_one(query, values={"id": user_id})
    return {"id": user_id}

# ✅ Regular sync function — also works fine
@app.get("/compute")
def heavy_computation(n: int):
    # CPU-bound: FastAPI runs this in a thread pool automatically
    result = sum(range(n))
    return {"result": result}

Rule of thumb:

  • Database queries, HTTP calls, file I/O → async def
  • CPU-heavy computation → regular def (FastAPI uses a thread pool)

Dependency Injection

FastAPI’s Depends() system makes it easy to share logic (auth, DB sessions, config) across endpoints.

from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Annotated

app = FastAPI()

# --- Dependency: get current user from token ---
def get_current_user(authorization: str | None = Header(None)):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Not authenticated")
    token = authorization[7:]
    # In production: validate JWT here
    return {"id": 1, "name": "Alice", "token": token}

CurrentUser = Annotated[dict, Depends(get_current_user)]

# --- Dependency: pagination params ---
def get_pagination(page: int = 1, size: int = 20) -> dict:
    if size > 100:
        raise HTTPException(status_code=400, detail="Max page size is 100")
    return {"skip": (page - 1) * size, "limit": size}

Pagination = Annotated[dict, Depends(get_pagination)]

# --- Using dependencies ---
@app.get("/me")
async def get_me(current_user: CurrentUser):
    return current_user

@app.get("/posts")
async def list_posts(current_user: CurrentUser, pagination: Pagination):
    # Only authenticated users can access this
    return {
        "user": current_user["name"],
        "pagination": pagination,
        "posts": []
    }

Database Integration — SQLAlchemy Async

The get_db function is implemented below. It handles the core logic described above:

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import Column, Integer, String, DateTime, select
from sqlalchemy.orm import DeclarativeBase
from datetime import datetime

# Database setup
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

# DB session dependency
async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user.id, "name": user.name, "email": user.email}

@app.post("/users/")
async def create_user(name: str, email: str, db: AsyncSession = Depends(get_db)):
    user = User(name=name, email=email)
    db.add(user)
    await db.flush()  # Get the auto-generated ID
    return {"id": user.id, "name": user.name}

JWT Authentication

The verify_password function is implemented below. It handles the core logic described above:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel

SECRET_KEY = "your-secret-key-min-32-chars"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    # In production: fetch user from DB here
    return {"id": int(user_id), "name": "Alice"}

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # In production: look up user in DB and verify password
    # user = await get_user(form_data.username)
    # if not verify_password(form_data.password, user.hashed_password):
    #     raise HTTPException(...)
    
    access_token = create_access_token(
        data={"sub": "1"},  # user ID
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/me")
async def read_me(current_user: dict = Depends(get_current_user)):
    return current_user

Background Tasks

The send_welcome_email function is implemented below. It handles the core logic described above:

from fastapi import FastAPI, BackgroundTasks
import smtplib

app = FastAPI()

def send_welcome_email(email: str, name: str):
    # Runs after the response is sent
    print(f"Sending welcome email to {email}...")
    # smtplib.SMTP(...).sendmail(...)

@app.post("/register")
async def register(
    email: str, 
    name: str, 
    background_tasks: BackgroundTasks
):
    # Immediately return a response
    background_tasks.add_task(send_welcome_email, email, name)
    return {"message": "Registration successful. Check your email."}

Error Handling

The __init__ function is implemented below. It handles the core logic described above:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# Custom exception
class AppError(Exception):
    def __init__(self, message: str, code: str = "ERROR"):
        self.message = message
        self.code = code

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(
        status_code=400,
        content={"error": exc.code, "message": exc.message}
    )

# Override validation error format
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(x) for x in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })
    return JSONResponse(status_code=422, content={"errors": errors})

@app.get("/risky")
async def risky_endpoint(fail: bool = False):
    if fail:
        raise AppError("Something went wrong", "OPERATION_FAILED")
    return {"status": "ok"}

Project Structure (Production)

myapi/
├── main.py              # App entry point
├── core/
│   ├── config.py        # Settings (pydantic-settings)
│   └── security.py      # JWT, password hashing
├── db/
│   ├── database.py      # Engine, session factory
│   └── models.py        # SQLAlchemy models
├── api/
│   ├── deps.py          # Shared dependencies
│   └── routes/
│       ├── users.py     # /users endpoints
│       ├── items.py     # /items endpoints
│       └── auth.py      # /token endpoint
├── schemas/
│   ├── user.py          # UserCreate, UserResponse, etc.
│   └── item.py
└── tests/
    ├── test_users.py
    └── conftest.py
# main.py — Router registration
from fastapi import FastAPI
from api.routes import users, items, auth

app = FastAPI(title="My API", version="1.0.0")

app.include_router(auth.router, prefix="/auth", tags=["auth"])
app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(items.router, prefix="/items", tags=["items"])

Deployment

Development

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Production

# Gunicorn + Uvicorn workers
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Docker

Configuration file:

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

Conclusion

FastAPI gives you production-grade API performance with minimal boilerplate. The key advantages:

  1. Auto docs — Swagger and ReDoc out of the box, always in sync with your code
  2. Pydantic validation — type errors are caught at startup
  3. Async by default — scale to more concurrent users without more hardware
  4. Dependency injection — clean, testable code architecture

Related posts: