FastAPI Tutorial: สร้าง REST API ด้วย Python ปี 2026

FastAPI has become the most popular Python web framework for building APIs, surpassing Flask in GitHub stars and adoption. It combines the simplicity of Flask with the performance of Node.js and Go, automatic interactive documentation, and built-in data validation using Python type hints. This comprehensive tutorial walks you through building production-ready REST APIs with FastAPI, covering everything from project setup to authentication, database integration, and deployment.

Why FastAPI?

FastAPI offers a unique combination of developer experience and performance that makes it the ideal choice for modern Python API development.

• Automatic interactive API documentation (Swagger UI + ReDoc)
• Data validation and serialization with Pydantic
• Native async/await support for high concurrency
• Type-safe development with Python type hints
• 40% fewer bugs through automatic validation (per FastAPI benchmarks)
• Performance on par with Node.js and Go frameworks

FastAPI vs Flask vs Django REST Framework

Project Setup

Set up a new FastAPI project with a proper project structure, virtual environment, and essential dependencies.

# Create project directory
mkdir fastapi-project && cd fastapi-project

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Linux/macOS
# venv\Scripts\activate   # Windows

# Install dependencies
pip install fastapi uvicorn[standard] pydantic-settings sqlalchemy[asyncio] asyncpg python-jose[cryptography] passlib[bcrypt] python-multipart httpx pytest

# Project structure
fastapi-project/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI app instance
│   ├── config.py          # Settings with pydantic-settings
│   ├── database.py        # Database connection
│   ├── models/            # SQLAlchemy models
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/           # Pydantic schemas
│   │   ├── __init__.py
│   │   └── user.py
│   ├── routers/           # API route handlers
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── auth.py
│   ├── services/          # Business logic
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── dependencies.py    # Shared dependencies
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_users.py
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

Your First API

Create a basic FastAPI application with a few endpoints to understand the fundamentals.

# app/main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize resources
    print("Starting up...")
    yield
    # Shutdown: cleanup resources
    print("Shutting down...")

app = FastAPI(
    title="My API",
    description="A production-ready REST API built with FastAPI",
    version="1.0.0",
    lifespan=lifespan,
)

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

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

# Run with:
# uvicorn app.main:app --reload --port 8000
#
# API docs available at:
# http://localhost:8000/docs      (Swagger UI)
# http://localhost:8000/redoc     (ReDoc)

Routing and Path Parameters

FastAPI supports path parameters, query parameters, and request bodies with automatic validation and documentation.

Path Parameters

from fastapi import FastAPI, Path, HTTPException

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int = Path(..., title="User ID", ge=1, description="The ID of the user")
):
    # user_id is automatically validated as a positive integer
    if user_id > 1000:
        raise HTTPException(status_code=404, detail="User not found")
    return {"user_id": user_id, "name": f"User {user_id}"}

# Enum path parameters
from enum import Enum

class UserRole(str, Enum):
    admin = "admin"
    editor = "editor"
    viewer = "viewer"

@app.get("/users/role/{role}")
async def get_users_by_role(role: UserRole):
    return {"role": role, "message": f"Listing all {role.value} users"}

Query Parameters

from fastapi import Query
from typing import Optional

@app.get("/users")
async def list_users(
    skip: int = Query(default=0, ge=0, description="Number of records to skip"),
    limit: int = Query(default=20, ge=1, le=100, description="Max records to return"),
    search: Optional[str] = Query(default=None, min_length=1, max_length=100),
    sort_by: str = Query(default="created_at", pattern="^(name|email|created_at)$"),
    active: bool = Query(default=True),
):
    return {
        "skip": skip,
        "limit": limit,
        "search": search,
        "sort_by": sort_by,
        "active": active,
    }

Request Body with Pydantic Models

Pydantic models provide automatic request validation, serialization, and documentation. This is one of the most powerful features of FastAPI.

# app/schemas/user.py
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from typing import Optional

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=100, examples=["John Doe"])
    email: EmailStr = Field(..., examples=["john@example.com"])
    password: str = Field(..., min_length=8, max_length=128)
    age: Optional[int] = Field(default=None, ge=13, le=150)
    bio: Optional[str] = Field(default=None, max_length=500)

    model_config = {
        "json_schema_extra": {
            "examples": [{
                "name": "John Doe",
                "email": "john@example.com",
                "password": "securePass123",
                "age": 28,
                "bio": "Software developer"
            }]
        }
    }

class UserResponse(BaseModel):
    id: int
    name: str
    email: EmailStr
    age: Optional[int] = None
    bio: Optional[str] = None
    created_at: datetime
    # Note: password is NOT included in the response

    model_config = {"from_attributes": True}

class UserUpdate(BaseModel):
    name: Optional[str] = Field(default=None, min_length=2, max_length=100)
    bio: Optional[str] = Field(default=None, max_length=500)
    age: Optional[int] = Field(default=None, ge=13, le=150)

# Usage in route
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    # Pydantic automatically validates the request body
    # Invalid data returns a 422 with detailed error messages
    return {"id": 1, **user.model_dump(), "created_at": datetime.now()}

Response Models

Use response models to control what data is returned to clients, filtering out sensitive fields and adding proper documentation.

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

class UserListResponse(BaseModel):
    users: List[UserResponse]
    total: int
    page: int
    per_page: int

@app.get("/users", response_model=UserListResponse)
async def list_users(page: int = 1, per_page: int = 20):
    # response_model ensures only declared fields are returned
    # Even if the database returns extra fields (like password_hash),
    # they will be filtered out automatically
    return {
        "users": [],
        "total": 0,
        "page": page,
        "per_page": per_page,
    }

Advanced Validation

Pydantic and FastAPI provide extensive validation capabilities using Field validators, custom validators, and constrained types.

from pydantic import BaseModel, field_validator, model_validator
from typing import Optional
import re

class UserCreate(BaseModel):
    username: str
    email: str
    password: str
    confirm_password: str

    @field_validator('username')
    @classmethod
    def username_must_be_alphanumeric(cls, v: str) -> str:
        if not re.match(r'^[a-zA-Z0-9_]+$', v):
            raise ValueError('Username must be alphanumeric')
        if len(v) < 3:
            raise ValueError('Username must be at least 3 characters')
        return v.lower()

    @field_validator('password')
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain an uppercase letter')
        if not re.search(r'[0-9]', v):
            raise ValueError('Password must contain a number')
        return v

    @model_validator(mode='after')
    def passwords_match(self):
        if self.password != self.confirm_password:
            raise ValueError('Passwords do not match')
        return self

Error Handling

FastAPI provides built-in exception handlers and allows you to create custom exception handlers for consistent error responses.

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# Custom exception class
class AppException(Exception):
    def __init__(self, status_code: int, detail: str, error_code: str):
        self.status_code = status_code
        self.detail = detail
        self.error_code = error_code

# Register custom exception handler
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.error_code,
            "detail": exc.detail,
            "path": str(request.url),
        },
    )

# Usage
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await find_user(user_id)
    if not user:
        raise AppException(
            status_code=404,
            detail=f"User with ID {user_id} not found",
            error_code="USER_NOT_FOUND",
        )
    return user

Database Integration

Connect FastAPI to a database using SQLAlchemy for SQL databases or Motor for MongoDB. This section covers the SQLAlchemy async setup.

# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

engine = create_async_engine(DATABASE_URL, echo=True, pool_size=20, max_overflow=10)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# Dependency for database sessions
async def get_db():
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# app/models/user.py
from sqlalchemy import Column, Integer, String, DateTime, func
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True, nullable=False, index=True)
    password_hash = Column(String(255), nullable=False)
    created_at = Column(DateTime, server_default=func.now())
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())

Complete CRUD API

Build a complete CRUD (Create, Read, Update, Delete) API with database integration.

# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.database import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserResponse, UserUpdate
from typing import List

router = APIRouter(prefix="/users", tags=["Users"])

@router.get("/", response_model=List[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 20,
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(User).offset(skip).limit(limit))
    users = result.scalars().all()
    return users

@router.get("/{user_id}", response_model=UserResponse)
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:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(data: UserCreate, db: AsyncSession = Depends(get_db)):
    user = User(name=data.name, email=data.email, password_hash=hash_password(data.password))
    db.add(user)
    await db.flush()
    await db.refresh(user)
    return user

@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(user_id: int, data: UserUpdate, 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:
        raise HTTPException(status_code=404, detail="User not found")
    for field, value in data.model_dump(exclude_unset=True).items():
        setattr(user, field, value)
    await db.flush()
    return user

@router.delete("/{user_id}", status_code=204)
async def delete_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:
        raise HTTPException(status_code=404, detail="User not found")
    await db.delete(user)

# Register router in main.py
# app.include_router(router)

Authentication and Authorization

Implement JWT-based authentication with OAuth2 password flow, which FastAPI supports out of the box with automatic documentation in Swagger UI.

# app/routers/auth.py
from fastapi import APIRouter, 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

router = APIRouter(prefix="/auth", tags=["Authentication"])

SECRET_KEY = "your-secret-key"  # Use environment variable in production
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

class Token(BaseModel):
    access_token: str
    token_type: str

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    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
    # Fetch user from database
    user = await find_user_by_id(int(user_id))
    if user is None:
        raise credentials_exception
    return user

@router.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = await authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
        )
    access_token = create_access_token(
        data={"sub": str(user.id)},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return {"access_token": access_token, "token_type": "bearer"}

# Protected endpoint
@router.get("/me", response_model=UserResponse)
async def get_current_user_profile(current_user = Depends(get_current_user)):
    return current_user

Middleware

Add middleware for logging, CORS, request timing, and other cross-cutting concerns.

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time, logging

app = FastAPI()

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# GZip compression
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Custom request timing middleware
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = f"{process_time:.4f}"
    logging.info(f"{request.method} {request.url.path} - {process_time:.4f}s")
    return response

Background Tasks

FastAPI supports background tasks that run after a response is sent, useful for sending emails, processing data, or other non-blocking operations.

from fastapi import BackgroundTasks

async def send_welcome_email(email: str, name: str):
    # Simulate sending email (replace with actual email service)
    await asyncio.sleep(2)
    print(f"Welcome email sent to {name} at {email}")

async def log_user_creation(user_id: int):
    print(f"User {user_id} created at {datetime.now()}")

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(
    data: UserCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
):
    user = await user_service.create(db, data)

    # These run AFTER the response is sent
    background_tasks.add_task(send_welcome_email, user.email, user.name)
    background_tasks.add_task(log_user_creation, user.id)

    return user  # Response is sent immediately

Testing

FastAPI is designed for easy testing. Use the TestClient (based on httpx) to test your API endpoints without starting a server.

# tests/test_users.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

@pytest.mark.anyio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users", json={
        "name": "John Doe",
        "email": "john@example.com",
        "password": "SecurePass123",
    })
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "John Doe"
    assert data["email"] == "john@example.com"
    assert "password" not in data  # Password should be excluded

@pytest.mark.anyio
async def test_create_user_validation_error(client: AsyncClient):
    response = await client.post("/users", json={
        "name": "J",  # Too short
        "email": "not-an-email",
        "password": "weak",
    })
    assert response.status_code == 422
    errors = response.json()["detail"]
    assert len(errors) > 0

@pytest.mark.anyio
async def test_get_user_not_found(client: AsyncClient):
    response = await client.get("/users/99999")
    assert response.status_code == 404

# Run tests:
# pytest tests/ -v --tb=short

Deployment

Deploy your FastAPI application to production using Docker and various cloud platforms.

Docker Deployment

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY ./app ./app

# Create non-root user
RUN adduser --disabled-password --no-create-home appuser
USER appuser

# Run with multiple workers for production
CMD ["uvicorn", "app.main:app", \
     "--host", "0.0.0.0", \
     "--port", "8000", \
     "--workers", "4"]

# docker-compose.yml
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/mydb
      - JWT_SECRET=your-production-secret
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:16
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - pg-data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user"]
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  pg-data:

Performance Tips

1. Use async endpoints for I/O-bound operations (database queries, HTTP requests).
2. Use sync endpoints for CPU-bound operations to avoid blocking the event loop.
3. Enable response compression with GZipMiddleware for large payloads.
4. Use connection pooling for database connections (SQLAlchemy async pool).
5. Implement caching with Redis for frequently accessed data.
6. Use multiple Uvicorn workers in production (or Gunicorn with Uvicorn workers).
7. Profile your application with py-spy or cProfile to find bottlenecks.

Best Practices

1. Use Pydantic models for all request/response schemas — never return raw dicts.
2. Separate your application into routers, models, schemas, and services.
3. Use dependency injection for database sessions, authentication, and shared logic.
4. Store configuration in environment variables using pydantic-settings.
5. Write comprehensive tests using TestClient and pytest fixtures.
6. Use async database drivers (asyncpg, aiosqlite) for async endpoints.
7. Document your API with descriptions, tags, and examples in your Pydantic models.
8. Handle errors consistently with custom exception handlers.

Conclusion

FastAPI is a production-ready framework that combines Python developer experience with high performance. Its automatic data validation through Pydantic, interactive API documentation, and native async support make it the best choice for building REST APIs in Python. Start with the project structure shown in this tutorial, define your Pydantic models first, then build your routes and business logic. Use dependency injection for database sessions and authentication, write tests from the beginning, and deploy with Docker for consistent environments. FastAPI seamless integration with the Python ecosystem means you can leverage thousands of existing libraries while building fast, type-safe, and well-documented APIs.

FAQ