🐍 Ngày 47 Python 365 ngày | Xây dựng API với FastAPI

🗓 Ngày 47: FastAPI — API hiện đại, nhanh và tự tài liệu hóa

Flask và Django đều ra đời trước kỷ nguyên async. FastAPI (ra mắt 2018) được thiết kế cho thế giới hiện đại: async/await, kiểm tra dữ liệu tự động bằng Pydantic, và tự sinh tài liệu API mà bạn không phải viết một dòng nào.

Nếu công việc của bạn là phục vụ model machine learning hoặc cung cấp API cho app di động, FastAPI là lựa chọn mặc định ngày nay.

---

🎯 Mục tiêu bài học

• Hiểu async/await và khi nào nó thực sự giúp ích.
• Tạo route GET/POST/PUT/DELETE đúng chuẩn REST.
• Validate dữ liệu bằng Pydantic model.
• Dùng path/query parameter có kiểu dữ liệu.
• Xử lý lỗi bằng HTTPException.
• Dùng Dependency Injection.
• Xây API phân tích crypto hoàn chỉnh có tài liệu tự sinh.

---

⚡ Vì sao FastAPI nhanh?

Yếu tố	Giải thích
ASGI + async	Xử lý nhiều request đồng thời trên một process, không chặn I/O
Pydantic (Rust)	Validate dữ liệu ở tốc độ native
Không magic	Ít lớp trừu tượng hơn Django
Type hints	Dùng chính annotation của Python làm schema

flowchart LR
    A[Request] --> B[Pydantic validate]
    B -->|Hợp lệ| C[Async handler]
    B -->|Sai| D[422 + chi tiết lỗi tự động]
    C --> E[Response model]
    E --> F[Tự sinh Swagger/OpenAPI]

---

🧰 Cài đặt

python -m venv .venv
.venv\Scripts\activate

pip install "fastapi[standard]" httpx

Chạy dev server:

fastapi dev main.py

Mở:

• http://127.0.0.1:8000/docs → Swagger UI (tự sinh!)
• http://127.0.0.1:8000/redoc → ReDoc

---

🚀 Bước 1: App nhỏ nhất

# main.py
from fastapi import FastAPI

app = FastAPI(
    title="API Phân tích Crypto",
    description="API demo cho chuỗi Python 365 ngày",
    version="1.0.0",
)


@app.get("/")
async def goc():
    return {"thong_diep": "Xin chào từ FastAPI 🚀"}


@app.get("/suc-khoe")
async def suc_khoe():
    return {"trang_thai": "ok"}

Chỉ cần chạy, bạn đã có tài liệu API tương tác đầy đủ. Đây là điểm "wow" lớn nhất của FastAPI.

---

📐 Bước 2: Path & Query parameter có kiểu

from fastapi import FastAPI, Query, Path
from typing import Annotated

app = FastAPI()


@app.get("/gia/{symbol}")
async def lay_gia(
    symbol: Annotated[str, Path(description="Cặp giao dịch", examples=["BTCUSDT"])],
    so_ngay: Annotated[int, Query(ge=1, le=365, description="Số ngày dữ liệu")] = 7,
):
    """Lấy giá của một cặp giao dịch."""
    return {"symbol": symbol.upper(), "so_ngay": so_ngay}

FastAPI tự động:

• Ép kiểu symbol thành str, so_ngay thành int.
• Trả 422 kèm giải thích nếu so_ngay=999 (vượt le=365).
• Hiển thị cả hai tham số trên Swagger UI.

💡 Quy ước: Tham số xuất hiện trong đường dẫn {...} là path param; tham số còn lại mặc định là query param. Muốn nhận body thì khai báo bằng Pydantic model.

---

✅ Bước 3: Pydantic — validate dữ liệu đầu vào

from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, Field, field_validator
from typing import Literal


class GiaoDichVao(BaseModel):
    """Dữ liệu client gửi lên khi tạo giao dịch."""

    symbol: str = Field(min_length=3, max_length=20, description="VD: BTCUSDT")
    loai: Literal["MUA", "BAN"]
    so_luong: Decimal = Field(gt=0, description="Phải lớn hơn 0")
    gia: Decimal = Field(gt=0)
    phi: Decimal = Field(default=Decimal("0"), ge=0)
    thoi_diem: datetime | None = None

    @field_validator("symbol")
    @classmethod
    def chuan_hoa_symbol(cls, v: str) -> str:
        v = v.strip().upper()
        if not v.isalnum():
            raise ValueError("Symbol chỉ được chứa chữ và số")
        return v


class GiaoDichRa(GiaoDichVao):
    """Dữ liệu trả về — thêm id và giá trị."""

    id: int
    gia_tri: Decimal

Dùng trong route:

@app.post("/giao-dich", response_model=GiaoDichRa, status_code=201)
async def tao_giao_dich(gd: GiaoDichVao):
    # Nếu code chạy tới đây, dữ liệu ĐÃ hợp lệ
    gia_tri = gd.so_luong * gd.gia + gd.phi
    return GiaoDichRa(id=1, gia_tri=gia_tri, **gd.model_dump())

Những gì FastAPI làm tự động cho bạn:

Kiểm tra	Kết quả khi sai
Thiếu trường bắt buộc	422 + tên trường thiếu
so_luong = -5	422 + "Input should be greater than 0"
loai = "XYZ"	422 + danh sách giá trị hợp lệ
symbol = "btc usdt"	422 + thông báo từ field_validator

Bạn không phải viết một dòng if nào.

---

🔀 Bước 4: CRUD đầy đủ theo chuẩn REST

from fastapi import HTTPException, status

# "Database" tạm
_db: dict[int, dict] = {}
_bo_dem = 0


@app.get("/giao-dich")
async def danh_sach(loai: Literal["MUA", "BAN"] | None = None, gioi_han: int = 50):
    ket_qua = list(_db.values())
    if loai:
        ket_qua = [g for g in ket_qua if g["loai"] == loai]
    return ket_qua[:gioi_han]


@app.get("/giao-dich/{gd_id}")
async def chi_tiet(gd_id: int):
    if gd_id not in _db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Không tìm thấy giao dịch id={gd_id}",
        )
    return _db[gd_id]


@app.post("/giao-dich", status_code=201)
async def tao_moi(gd: GiaoDichVao):
    global _bo_dem
    _bo_dem += 1
    _db[_bo_dem] = {"id": _bo_dem, **gd.model_dump()}
    return _db[_bo_dem]


@app.put("/giao-dich/{gd_id}")
async def cap_nhat(gd_id: int, gd: GiaoDichVao):
    if gd_id not in _db:
        raise HTTPException(404, "Không tìm thấy giao dịch")
    _db[gd_id] = {"id": gd_id, **gd.model_dump()}
    return _db[gd_id]


@app.delete("/giao-dich/{gd_id}", status_code=204)
async def xoa(gd_id: int):
    if _db.pop(gd_id, None) is None:
        raise HTTPException(404, "Không tìm thấy giao dịch")
    return None

Bảng mã trạng thái cần nhớ:

Method	Đường dẫn	Thành công	Ý nghĩa
GET	/giao-dich	200	Danh sách
GET	/giao-dich/{id}	200	Một bản ghi
POST	/giao-dich	201	Tạo mới
PUT	/giao-dich/{id}	200	Cập nhật toàn bộ
PATCH	/giao-dich/{id}	200	Cập nhật một phần
DELETE	/giao-dich/{id}	204	Xóa (không có body)

---

🔄 Bước 5: Async thật sự — gọi API đồng thời

Đây là nơi async trả cổ tức. So sánh 3 cách gọi 3 API:

import asyncio
import httpx

COINS = ["bitcoin", "ethereum", "solana"]
URL = "https://api.coingecko.com/api/v3/simple/price"

# ❌ CÁCH CHẬM: tuần tự — ~3 giây (mỗi lần 1s)
def lay_tuan_tu():
    ket_qua = {}
    with httpx.Client(timeout=10) as c:
        for coin in COINS:
            r = c.get(URL, params={"ids": coin, "vs_currencies": "usd"})
            ket_qua[coin] = r.json()[coin]["usd"]
    return ket_qua

# ✅ CÁCH NHANH: đồng thời — ~1 giây (chờ lâu nhất)
async def lay_dong_thoi():
    async with httpx.AsyncClient(timeout=10) as c:
        tac_vu = [
            c.get(URL, params={"ids": coin, "vs_currencies": "usd"})
            for coin in COINS
        ]
        phan_hoi = await asyncio.gather(*tac_vu)
        return {coin: r.json()[coin]["usd"] for coin, r in zip(COINS, phan_hoi)}


@app.get("/gia-dong-thoi")
async def gia_dong_thoi():
    return await lay_dong_thoi()

Ba khái niệm cần nắm:

Khái niệm	Ý nghĩa
await	Nhường quyền điều khiển trong lúc chờ I/O
asyncio.gather()	Chạy nhiều coroutine đồng thời
httpx.AsyncClient	HTTP client hỗ trợ async (requests không hỗ trợ)

⚠️ Bẫy lớn nhất: gọi code blocking (như requests.get, time.sleep, truy vấn DB đồng bộ) bên trong async def sẽ chặn toàn bộ server. Nếu buộc phải dùng thư viện blocking, hãy khai báo route bằng def (không async) — FastAPI sẽ tự chạy nó trong threadpool.

---

💉 Bước 6: Dependency Injection

from fastapi import Depends, Header
from typing import Annotated


async def lay_api_key(x_api_key: Annotated[str | None, Header()] = None) -> str:
    if not x_api_key:
        raise HTTPException(401, "Thiếu header X-API-Key")
    if x_api_key != "demo-key-123":
        raise HTTPException(403, "API key không hợp lệ")
    return x_api_key


async def lay_phan_trang(skip: int = 0, limit: Annotated[int, Query(le=100)] = 20):
    return {"skip": skip, "limit": limit}


@app.get("/bao-mat/du-lieu")
async def du_lieu_bao_mat(
    api_key: Annotated[str, Depends(lay_api_key)],
    trang: Annotated[dict, Depends(lay_phan_trang)],
):
    return {"da_xac_thuc": True, "trang": trang}

Lợi ích: viết logic một lần, tái sử dụng ở hàng chục route; FastAPI cache kết quả dependency trong cùng một request.

---

🏗️ Bước 7: API hoàn chỉnh + cấu trúc dự án

crypto_api/
├── main.py
├── models.py
├── services.py
└── routers/
    ├── giao_dich.py
    └── thi_truong.py

services.py:

import httpx
from decimal import Decimal

COINGECKO = "https://api.coingecko.com/api/v3/simple/price"
BANG_DOI = {"BTCUSDT": "bitcoin", "ETHUSDT": "ethereum", "SOLUSDT": "solana"}


async def lay_gia(symbol: str) -> dict:
    coin = BANG_DOI.get(symbol.upper())
    if not coin:
        return {"loi": f"Chưa hỗ trợ {symbol}"}
    async with httpx.AsyncClient(timeout=10) as c:
        r = await c.get(COINGECKO, params={
            "ids": coin, "vs_currencies": "usd", "include_24hr_change": "true",
        })
        r.raise_for_status()
        d = r.json()[coin]
    return {"symbol": symbol.upper(), "gia_usd": Decimal(str(d["usd"])),
            "thay_doi_24h": round(d.get("usd_24h_change", 0), 2)}

routers/thi_truong.py:

from fastapi import APIRouter, HTTPException
from services import lay_gia

router = APIRouter(prefix="/thi-truong", tags=["Thị trường"])


@router.get("/{symbol}")
async def gia(symbol: str):
    kq = await lay_gia(symbol)
    if "loi" in kq:
        raise HTTPException(404, kq["loi"])
    return kq

main.py:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from routers import thi_truong, giao_dich

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

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://www.huongnghieppython.com"],
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(thi_truong.router)
app.include_router(giao_dich.router)

---

🚢 Bước 8: Deploy

pip freeze > requirements.txt
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

Nền tảng	Ghi chú
Vercel	Hỗ trợ ASGI; phù hợp API nhỏ
Render / Railway	Push Git → tự deploy
Docker	FROM python:3.12-slim + uvicorn

Checklist production:

• --reload chỉ dùng khi dev
• Cấu hình CORS chính xác (không dùng allow_origins=["*"] ở production)
• Giới hạn rate limit (dùng slowapi)
• Đặt sau reverse proxy (nginx) với HTTPS
• Log và health check (/suc-khoe)

---

🧪 Bài tập thực hành

1. Thêm endpoint /thi-truong/{symbol}/lich-su?so_ngay=30 trả dữ liệu nến.
2. Viết response_model để ẩn trường phi khi trả về.
3. Thêm dependency kiểm tra quyền admin.
4. Dùng BackgroundTasks để ghi log bất đồng bộ.
5. So sánh thời gian /gia-tuan-tu và /gia-dong-thoi bằng time.perf_counter().
6. Thêm unit test với TestClient.

---

⚠️ Lỗi thường gặp

Lỗi	Nguyên nhân	Cách sửa
422 Unprocessable Entity	Body không khớp model	Đọc chi tiết lỗi trong response
Server "đứng" khi nhiều request	Dùng requests/time.sleep trong async def	Đổi sang httpx.AsyncClient/asyncio.sleep, hoặc bỏ async
RuntimeError: no running event loop	Gọi coroutine mà không await	Thêm await
CORS policy trên browser	Chưa thêm middleware	app.add_middleware(CORSMiddleware, ...)
Swagger không hiện route	Quên include_router	Kiểm tra app.include_router(...)
Decimal không serialize	Pydantic v1 vs v2 khác nhau	Dùng Decimal + model_config phù hợp

---

📝 Ghi chú

• FastAPI không có ORM. Dùng SQLModel (của cùng tác giả, gộp Pydantic + SQLAlchemy) hoặc SQLAlchemy 2.0 async.
• Type hints không phải trang trí — chúng là schema API. Sai type = sai tài liệu = bug.
• response_model bảo vệ bạn: nó lọc bỏ những trường bạn vô tình trả ra (ví dụ mật khẩu).
• Tự động sinh OpenAPI nghĩa là bạn có thể sinh client SDK cho mobile/web chỉ từ một URL.
• Với model ML: FastAPI + BackgroundTasks + Redis là combo rất phổ biến.

---

🎯 Tóm tắt ngày 47

✅ Hiểu async/await và tránh bẫy blocking ✅ Xây CRUD REST đúng chuẩn với mã trạng thái chính xác ✅ Validate dữ liệu hoàn toàn bằng Pydantic ✅ Gọi nhiều API đồng thời với asyncio.gather ✅ Dùng Dependency Injection tái sử dụng logic ✅ Tổ chức dự án với APIRouter + deploy bằng uvicorn

---

📚 Học tiếp

• ⬅️ Ngày 46 — Xây dựng ứng dụng web với Django
• ➡️ Ngày 48 — Xây dựng ứng dụng web với Streamlit
• 📘 Lộ trình học Python FastAPI Backend
• 📘 Python Automation — Tự động hóa công việc với Python