Data API 2 — FastAPI căn bản

13 thg 7, 2026 3 lượt xem
#python
#async
#fastapi
#openapi
#pydantic

Sau bài tổng quan REST API, giờ ta bắt tay vào framework cụ thể. FastAPI là lựa chọn mặc định của nhiều team dữ liệu Việt Nam khi cần dựng một service HTTP: nhanh, gõ ít code, và quan trọng nhất với dân data — nó tự sinh tài liệu API và tự validate dữ liệu đầu vào dựa trên type hint. Bài này đi qua toàn bộ nền tảng để bạn có thể tự tin viết một endpoint hoàn chỉnh trước khi qua các bài chuyên sâu về validation, database và bảo mật.

1. Cài đặt và khởi tạo

FastAPI là thư viện ASGI, cần một máy chủ ASGI để chạy. Bộ đôi phổ biến là fastapi + uvicorn.

pip install "fastapi[standard]" uvicorn

File tối thiểu main.py:

from fastapi import FastAPI

app = FastAPI(
    title="NCB Data API",
    description="Dịch vụ tra cứu dữ liệu giao dịch cho hệ thống nội bộ",
    version="1.0.0",
)


@app.get("/health")
def health():
    return {"status": "ok"}

Chạy môi trường dev với reload tự động khi sửa code:

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

Cú pháp main:app nghĩa là "trong module main, lấy đối tượng tên app". Cờ --reload chỉ dùng cho dev — không bao giờ bật ở production (ta sẽ bàn ở bài triển khai). Đối tượng FastAPI() là trung tâm: mọi route, middleware, cấu hình OpenAPI đều gắn vào nó.

2. Path operation và decorator

Một path operation là cặp (đường dẫn, phương thức HTTP) ánh xạ tới một hàm xử lý. FastAPI dùng decorator để khai báo:

DecoratorÝ nghĩa
@app.get(path)Đọc dữ liệu (idempotent)
@app.post(path)Tạo tài nguyên mới
@app.put(path)Thay thế toàn bộ tài nguyên
@app.patch(path)Cập nhật một phần
@app.delete(path)Xóa tài nguyên
@app.get("/accounts")
def list_accounts():
    return [{"id": 1, "account_no": "0011000123456"}]

Hàm bên dưới decorator gọi là path operation function. Giá trị nó return sẽ được FastAPI serialize thành JSON. Bạn trả về dict, list, model Pydantic, hay kiểu cơ bản — framework tự chuyển đổi.

3. Path parameter và query parameter

Path parameter

Tham số nằm trong chính đường dẫn, khai báo bằng dấu ngoặc nhọn. Khai báo type hint để FastAPI tự ép kiểu và validate:

@app.get("/accounts/{account_id}")
def get_account(account_id: int):
    return {"account_id": account_id}

Gọi /accounts/42 → account_id = 42 (int). Gọi /accounts/abc → FastAPI tự trả về lỗi 422 Unprocessable Entity kèm mô tả rõ ràng, hàm của bạn không bao giờ nhận rác.

Query parameter

Mọi tham số của hàm không khớp với path parameter sẽ được hiểu là query parameter (phần sau dấu ? trên URL):

from typing import Optional


@app.get("/transactions")
def search_transactions(
    limit: int = 20,              # có mặc định → optional
    offset: int = 0,
    kind: Optional[str] = None,   # có thể vắng mặt
):
    return {"limit": limit, "offset": offset, "kind": kind}

Quy tắc:

  • Có giá trị mặc định → tham số tùy chọn. /transactions dùng limit=20.
  • Không có mặc định → bắt buộc. Thiếu thì 422.
  • Optional[str] = None (tương đương str | None = None) → cho phép vắng mặt và mang giá trị None.

Để ràng buộc chặt hơn (khoảng giá trị, độ dài, regex), dùng Query:

from fastapi import Query


@app.get("/transactions")
def search(limit: int = Query(default=20, ge=1, le=100)):
    return {"limit": limit}

ge=1, le=100 đảm bảo limit nằm trong [1, 100] — chặn ngay các query "kéo cả bảng" gây quá tải.

4. Request body với Pydantic BaseModel

Với POST/PUT, dữ liệu đi trong body. FastAPI dùng Pydantic để mô tả và validate body. Khai báo một BaseModel và dùng nó làm type hint:

from pydantic import BaseModel, Field


class TransferRequest(BaseModel):
    from_account: str = Field(..., min_length=10, max_length=20)
    to_account: str = Field(..., min_length=10, max_length=20)
    amount: float = Field(..., gt=0)
    note: str | None = None


@app.post("/transfers")
def create_transfer(payload: TransferRequest):
    return {"received": payload.amount}

Khi request tới, FastAPI: đọc JSON body → validate theo TransferRequest → nếu sai kiểu/thiếu trường/vi phạm ràng buộc thì trả 422 với chi tiết; nếu hợp lệ, đưa vào hàm một object payload đã được ép kiểu sẵn. Dấu ... trong Field(...) nghĩa là trường bắt buộc. Chi tiết sâu về validation nằm ở bài validation dữ liệu.

5. response_model và status_code

Bạn nên khai báo rõ kiểu dữ liệu trả về bằng response_model. Lợi ích: (1) lọc bỏ các field không nên lộ ra (ví dụ số dư nội bộ), (2) tài liệu OpenAPI mô tả chính xác output, (3) FastAPI tự validate cả output.

from fastapi import status


class AccountOut(BaseModel):
    id: int
    account_no: str
    currency: str
    # KHÔNG khai báo balance → sẽ bị loại khỏi response


@app.post(
    "/accounts",
    response_model=AccountOut,
    status_code=status.HTTP_201_CREATED,
)
def create_account(payload: TransferRequest):
    fake = {"id": 1, "account_no": "0011...", "currency": "VND", "balance": 999}
    return fake  # balance tự động bị lược bỏ nhờ response_model

status_code=201 báo đúng ngữ nghĩa "đã tạo tài nguyên". Mặc định GET trả 200.

6. async def hay def?

Đây là điểm hay gây bối rối. FastAPI hỗ trợ cả hai kiểu hàm handler:

  • async def: dùng khi bên trong bạn gọi thư viện bất đồng bộ (await), ví dụ httpx.AsyncClient, driver DB async như asyncpg, databases. Handler chạy trực tiếp trên event loop.
  • def (thường): dùng khi code là đồng bộ chặn (blocking) — psycopg2, requests, pandas, đọc file. FastAPI tự động chạy các hàm def trong một threadpool riêng, nên chúng không chặn event loop.
import httpx


@app.get("/rate")
async def get_rate():
    async with httpx.AsyncClient() as client:      # gọi async → await
        r = await client.get("https://example/fx")
        return r.json()


@app.get("/report")
def build_report():
    df = read_big_csv()   # pandas blocking → để def, chạy trong threadpool
    return {"rows": len(df)}

Sai lầm nguy hiểm nhất: đặt code blocking bên trong async def. Khi đó nó chạy thẳng trên event loop và chặn toàn bộ server — mọi request khác phải chờ. Quy tắc an toàn: nếu không có await nào trong hàm, hãy dùng def thường. Chủ đề tối ưu async/performance có bài riêng và liên hệ với SQL, API & Async trong Python.

7. Tài liệu OpenAPI tự sinh

Đây là điểm mạnh lớn nhất của FastAPI. Từ type hint và Pydantic model, nó tự sinh OpenAPI schema (JSON tại /openapi.json) và hai giao diện tài liệu tương tác:

  • Swagger UI tại /docs — cho phép gõ tham số và bấm "Try it out" gọi thử API ngay trên trình duyệt.
  • ReDoc tại /redoc — giao diện đọc, phù hợp làm tài liệu chia sẻ.

Không cần viết thêm dòng nào. Với data team, nghĩa là cán bộ nghiệp vụ có thể tự thử API mà không cần bạn giải thích từng tham số.

Làm tài liệu đẹp hơn

Bạn có thể bổ sung tags, summary, mô tả và ví dụ cho từng endpoint:

@app.get(
    "/accounts/{account_id}/transactions",
    tags=["transactions"],
    summary="Lịch sử giao dịch của một tài khoản",
    response_description="Danh sách giao dịch đã phân trang",
)
def list_txn(account_id: int):
    ...

tags gom các endpoint thành nhóm trong Swagger UI. summary là dòng tiêu đề ngắn. Docstring của hàm trở thành mô tả chi tiết (hỗ trợ Markdown). Ta có thể gắn examples vào field Pydantic để tài liệu hiển thị mẫu dữ liệu thật:

class TransferRequest(BaseModel):
    amount: float = Field(..., gt=0, examples=[500000])

Bài kiểm thử và tài liệu đi sâu hơn về phần này.

8. APIRouter — chia module

Khi service lớn dần, dồn mọi route vào main.py sẽ rối. APIRouter cho phép tách route theo domain rồi gắn vào app chính:

# routers/transactions.py
from fastapi import APIRouter

router = APIRouter(prefix="/accounts", tags=["transactions"])


@router.get("/{account_id}/transactions")
def list_txn(account_id: int):
    return []

# main.py
from fastapi import FastAPI
from routers import transactions

app = FastAPI()
app.include_router(transactions.router)

prefix được thêm vào mọi đường dẫn trong router, tags áp cho cả nhóm. Nhờ vậy mỗi domain (accounts, transactions, customers) nằm ở một file riêng, dễ bảo trì — tinh thần tách trách nhiệm giống OOP & clean code.

9. Xử lý lỗi với HTTPException

Khi muốn trả lỗi HTTP có ngữ nghĩa (không tìm thấy, không có quyền...), raise HTTPException:

from fastapi import HTTPException, status


@app.get("/accounts/{account_id}")
def get_account(account_id: int):
    account = db_lookup(account_id)   # minh hoạ
    if account is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Không tìm thấy tài khoản {account_id}",
        )
    return account

FastAPI bắt exception này và trả về {"detail": "..."} với đúng status code. Với logic phức tạp hơn (custom exception handler, response lỗi thống nhất) sẽ bàn ở các bài sau. Phần xác thực/phân quyền nằm ở bài auth & bảo mật.

10. Cấu trúc project gợi ý

Bố cục thư mục thực dụng cho một data service cỡ vừa:

ncb-data-api/
├── app/
│   ├── main.py            # tạo FastAPI(), include_router
│   ├── routers/
│   │   ├── accounts.py
│   │   └── transactions.py
│   ├── schemas/           # Pydantic BaseModel (in/out)
│   │   └── transaction.py
│   ├── services/          # logic nghiệp vụ, truy vấn DB
│   └── core/              # config, kết nối DB, dependency
└── tests/

Tách schemas (hình dạng dữ liệu), services (nghiệp vụ), routers (giao thức HTTP) giúp mỗi lớp thay đổi độc lập — DB đổi không kéo theo sửa API. Việc tích hợp DB thật là nội dung bài database integration.

Use case thực tế

Bài toán: Ứng dụng Internet Banking của NCB cần một endpoint trả lịch sử giao dịch của một tài khoản, có phân trang để màn hình chỉ tải 20 giao dịch mỗi lần thay vì kéo cả chục nghìn dòng.

Thiết kế: GET /accounts/{account_id}/transactions?limit=20&offset=0

  • account_id: path parameter (bắt buộc, kiểu int).
  • limit: query parameter, mặc định 20, ràng buộc 1–100 để tránh quá tải.
  • offset: query parameter, mặc định 0, dùng để nhảy trang.
  • Trả về response_model chứa metadata phân trang + danh sách giao dịch.

Endpoint hoàn chỉnh (minh hoạ — không kết nối DB thật):

from datetime import datetime
from fastapi import APIRouter, HTTPException, Query, status
from pydantic import BaseModel

router = APIRouter(prefix="/accounts", tags=["transactions"])


class TransactionOut(BaseModel):
    id: int
    amount: float
    kind: str
    created_at: datetime


class TransactionPage(BaseModel):
    account_id: int
    total: int          # tổng số giao dịch của tài khoản
    limit: int
    offset: int
    items: list[TransactionOut]


@router.get(
    "/{account_id}/transactions",
    response_model=TransactionPage,
    summary="Lịch sử giao dịch của một tài khoản (có phân trang)",
    response_description="Trang giao dịch kèm metadata phân trang",
)
def list_transactions(
    account_id: int,
    limit: int = Query(default=20, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
):
    """
    Trả về lịch sử giao dịch của `account_id`, sắp xếp mới nhất trước.
    Dùng `limit`/`offset` để phân trang.
    """
    account = fake_get_account(account_id)          # minh hoạ
    if account is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Không tìm thấy tài khoản {account_id}",
        )

    total = fake_count_transactions(account_id)     # minh hoạ
    rows = fake_query_transactions(                 # minh hoạ
        account_id=account_id, limit=limit, offset=offset
    )

    return TransactionPage(
        account_id=account_id,
        total=total,
        limit=limit,
        offset=offset,
        items=rows,
    )

Về mặt truy vấn, phần lõi khi nối DB thật (PostgreSQL) tương đương câu SQL sau — đây đúng là loại truy vấn chạy được trên sandbox chỉ-đọc của ta:

-- ▶ Chạy được
SELECT id, amount, kind, created_at
FROM transactions
WHERE account_id = 1
ORDER BY created_at DESC
LIMIT 20 OFFSET 0;

Kết quả: màn hình lịch sử tải nhanh, mỗi lần chỉ trả ~20 dòng; client biết total để hiển thị "Trang 1/N"; ràng buộc limit ≤ 100 chặn các request cố kéo toàn bộ dữ liệu. Cùng một endpoint phục vụ được cả app di động lẫn màn hình quầy giao dịch.

Ghi nhớ

  • app = FastAPI() là trung tâm; chạy dev bằng uvicorn main:app --reload (không dùng --reload ở production).
  • Decorator @app.get/post/put/patch/delete khai báo path operation; giá trị return được serialize thành JSON.
  • Path parameter nằm trong {} của đường dẫn; query parameter là tham số hàm không khớp path — có mặc định thì tùy chọn, không có thì bắt buộc.
  • Request body mô tả bằng Pydantic BaseModel; FastAPI tự validate và trả 422 khi sai — không cần viết code kiểm tra thủ công.
  • Luôn khai báo response_model để lọc field nhạy cảm và tài liệu hóa output chính xác; đặt status_code đúng ngữ nghĩa (ví dụ 201 khi tạo).
  • Dùng async def khi có await; dùng def thường cho code blocking (nó chạy trong threadpool). Không đặt code blocking trong async def.
  • OpenAPI/Swagger UI (/docs) và ReDoc (/redoc) sinh tự động — điểm mạnh lớn giúp người dùng tự thử API.
  • APIRouter tách route theo domain; HTTPException trả lỗi HTTP có ngữ nghĩa (ví dụ 404).
  • Phân trang bằng limit/offset với ràng buộc khoảng giá trị là cách chuẩn để tránh kéo toàn bộ dữ liệu.

Bài viết liên quan

Python
Python 6 — Xử lý lỗi, File I/O & Logging

Exception handling, context manager (with), đọc/ghi file, JSON/CSV và logging đúng cách.

13 thg 7, 2026 4
Python
Python 4 — Hàm, Module & Môi trường ảo

Định nghĩa hàm, tham số, *args/**kwargs, lambda, module/package, pip và virtualenv.

13 thg 7, 2026 4
Python
Python 5 — OOP & Clean Code

Lớp, kế thừa, đa hình, dunder methods, dataclass, type hints và nguyên tắc viết code sạch.

13 thg 7, 2026 4
Python
Python 8 — Làm việc với SQL, API & Async

Kết nối CSDL với SQLAlchemy, gọi REST API với requests/httpx, và lập trình bất đồng bộ asyncio.

13 thg 7, 2026 4