FastAPI là một khung làm việc hiện đại, mạnh mẽ và nhanh chóng để xây dựng API bằng Python. Nhờ tích hợp sẵn hỗ trợ bất đồng bộ, kiểm tra kiểu dữ liệu tự động và tài liệu hóa API theo chuẩn OpenAPI, FastAPI giúp phát triển ứng dụng web với tốc độ cao và độ tin cậy cao.

1. Thiết lập môi trường ban đầu

Để bắt đầu, cần cài đặt hai thành phần nền tảng:

pip install fastapi uvicorn[standard]

Trong đó:

• FastAPI là framework xử lý logic yêu cầu và phản hồi.
• Uvicorn là máy chủ ASGI hiệu suất cao, hỗ trợ chạy ứng dụng FastAPI dưới mô hình bất đồng bộ.

Một cấu trúc thư mục tiêu chuẩn cho dự án có thể như sau:

fastapi-demo/
├── api/
│   ├── __init__.py
│   └── endpoints.py
├── core/
│   ├── __init__.py
│   └── database.py
├── schemas/
│   ├── __init__.py
│   └── item.py
└── main.py

Tệp main.py chứa điểm vào ứng dụng:

# main.py
from fastapi import FastAPI

app = FastAPI(title="Demo API", version="0.1.0")

@app.get("/")
def home():
    return {"status": "running", "framework": "FastAPI"}

Chạy lệnh sau để khởi động máy chủ với chế độ giám sát thay đổi mã nguồn:

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

Sau khi chạy, truy cập http://localhost:8000/docs để xem tài liệu API tương tác được tạo tự động — không cần viết thêm bất kỳ dòng nào.

2. Định tuyến và xử lý yêu cầu thông minh

FastAPI sử dụng decorator để gắn hàm xử lý với phương thức HTTP và đường dẫn cụ thể. Hệ thống tự động suy luận loại dữ liệu từ chú thích kiểu (type hints), đảm bảo tính đúng đắn ngay từ đầu.

Ví dụ định nghĩa endpoint lấy thông tin sản phẩm theo ID:

# api/endpoints.py
from fastapi import APIRouter
from schemas.item import ProductSchema

router = APIRouter()

@router.get("/products/{pid}", response_model=ProductSchema)
def fetch_product(pid: int, include_details: bool = False):
    # Giả lập truy vấn cơ sở dữ liệu
    return ProductSchema(
        id=pid,
        name=f"Product-{pid}",
        price=99.99 if include_details else None
    )

Ở đây, pid: int đảm bảo giá trị đường dẫn luôn là số nguyên; nếu gửi /products/abc, FastAPI trả về lỗi 422 với mô tả rõ ràng. Tham số include_details là query parameter tùy chọn, mặc định là False.

Để nhận dữ liệu JSON trong yêu cầu POST, ta dùng Pydantic model:

# schemas/item.py
from pydantic import BaseModel
from typing import Optional

class ProductSchema(BaseModel):
    id: int
    name: str
    price: float
    category: Optional[str] = None

Khi gọi POST /products/ với dữ liệu JSON hợp lệ, FastAPI tự động chuyển đổi sang đối tượng ProductSchema và kiểm tra toàn bộ ràng buộc kiểu và giá trị.

3. Tích hợp cơ sở dữ liệu với SQLAlchemy

Để làm việc với cơ sở dữ liệu quan hệ, ta kết nối qua SQLAlchemy ORM. Ví dụ cấu hình SQLite trong core/database.py:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./data/app.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

Mô hình dữ liệu đơn giản:

# core/models.py
from sqlalchemy import Column, Integer, String, Float
from .database import Base

class ProductDB(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    price = Column(Float)
    category = Column(String, nullable=True)

Khởi tạo bảng và cung cấp dependency injection cho session:

# core/database.py
def get_session():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Sử dụng trong endpoint:

# api/endpoints.py
from fastapi import Depends
from sqlalchemy.orm import Session
from core.database import get_session
from core.models import ProductDB

@router.post("/products/", response_model=ProductSchema)
def add_product(payload: ProductSchema, db: Session = Depends(get_session)):
    new_item = ProductDB(**payload.dict())
    db.add(new_item)
    db.commit()
    db.refresh(new_item)
    return new_item

4. Tối ưu hiệu năng ứng dụng

FastAPI tận dụng tối đa khả năng bất đồng bộ của Python. Các endpoint có thể được khai báo dưới dạng hàm async để xử lý song song các tác vụ I/O:

@router.get("/products/async/{pid}")
async def fetch_async(pid: int):
    # Giả lập gọi dịch vụ bên ngoài hoặc truy vấn DB bất đồng bộ
    await asyncio.sleep(0.1)  # thay bằng database.fetch_one(...) thực tế
    return {"id": pid, "fetched": "asynchronously"}

Để ghi log chi tiết, tích hợp module logging chuẩn của Python:

import logging
from fastapi import Request

logger = logging.getLogger("app.access")
logger.setLevel(logging.INFO)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"{request.method} {request.url.path}")
    response = await call_next(request)
    logger.info(f"Status: {response.status_code}")
    return response

Các kỹ thuật nâng cao khác bao gồm: sử dụng cache cho dữ liệu tĩnh, phân trang với offset/limit, tối ưu truy vấn SQL bằng selectinload, và triển khai background tasks cho các công việc tốn thời gian.