Data API 4 — Kết nối cơ sở dữ liệu

Ba bài trước dựng khung FastAPI và validate dữ liệu. Nhưng một data API thực chỉ có giá trị khi nó đọc dữ liệu thật — ở đây là Postgres chứa bảng customers, accounts, transactions. Bài này tập trung vào tầng nối giữa endpoint và database: làm sao mở kết nối đúng cách, dùng lại kết nối qua pool, truy vấn an toàn, phân trang cho bảng hàng chục triệu dòng, và biến lỗi DB thành mã HTTP hợp lý. Bối cảnh xuyên suốt là endpoint tra cứu lịch sử giao dịch và số dư của khách hàng.

1. Vì sao không "mở kết nối trong mỗi hàm"

Cách sai kinh điển là mỗi endpoint tự psycopg2.connect(...) rồi close(). Mỗi kết nối tới Postgres tốn một quy trình bắt tay TCP, xác thực, cấp một backend process phía server — cỡ vài mili-giây đến hàng chục ms. Với API tải cao (hàng nghìn request/giây khi phòng giao dịch tra cứu đồng loạt), chi phí này giết chết throughput và làm cạn max_connections của Postgres (mặc định thường 100).

Giải pháp là connection pool: một tập kết nối được mở sẵn, tái sử dụng. Request mượn một kết nối, dùng xong trả lại pool thay vì đóng. Đây là lý do ta luôn tạo một engine/pool ở cấp ứng dụng, chia sẻ cho toàn bộ request.

2. SQLAlchemy 2.0: Core, ORM và driver async

SQLAlchemy là lớp trừu tượng DB phổ biến nhất trong Python. Phiên bản 2.0 hợp nhất API và hỗ trợ async gốc.

• Core: viết truy vấn bằng biểu thức Python gần với SQL (select(...)), trả về row-like. Nhẹ, gần SQL, phù hợp data API chỉ đọc.
• ORM: ánh xạ hàng thành object Python (class Transaction). Tiện cho logic nghiệp vụ phức tạp, nhưng thêm chi phí và độ ma thuật.

Với một service tra cứu chủ yếu SELECT, Core (hoặc thậm chí SQL thô tham số hoá) thường là lựa chọn sáng suốt: dễ đọc, dễ tối ưu, ít bất ngờ.

Về driver, có hai chế độ:

asyncpg là driver async thuần, rất nhanh. Chuỗi kết nối tương ứng là postgresql+asyncpg://user:pass@host:5432/db. Xem thêm nền tảng ở bài Python SQL/API/async.

3. Tạo engine với cấu hình pool

Engine được tạo một lần khi ứng dụng khởi động. Các tham số pool quyết định hành vi dưới tải:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://api_ro:***@db-host:5432/core",
    pool_size=10,        # số kết nối giữ thường trực
    max_overflow=20,     # kết nối tạm được mở thêm khi pool cạn
    pool_timeout=30,     # giây chờ mượn conn trước khi báo lỗi
    pool_recycle=1800,   # tái tạo conn sau 30 phút, tránh conn "chết"
    pool_pre_ping=True,  # ping trước khi dùng, tránh conn đã bị server đóng
)

SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

Ý nghĩa và cách chọn:

• pool_size: số kết nối thường trực. Không phải càng lớn càng tốt — tổng pool_size của tất cả instance API phải nhỏ hơn max_connections của Postgres. Ví dụ 4 instance × (10 + 20) = 120 đã vượt mức 100 mặc định, cần chỉnh Postgres hoặc dùng PgBouncer ở giữa.
• max_overflow: kết nối "mượn thêm" khi burst, tự đóng khi rảnh. Đặt pool_size vừa đủ cho tải nền, max_overflow để chịu đỉnh.
• pool_pre_ping: rất nên bật — Postgres/firewall có thể đóng conn nhàn rỗi, pre-ping loại bỏ conn chết trước khi giao cho request, tránh lỗi ngẫu nhiên.
• pool_recycle: chống conn bị hạ ngầm sau thời gian dài.

Quy tắc thực chiến: bắt đầu pool_size ≈ số worker × 2, đo p95 latency và tỉ lệ pool_timeout, rồi tinh chỉnh. Nếu thấy nhiều TimeoutError khi mượn conn → hoặc query chậm giữ conn quá lâu, hoặc pool quá nhỏ.

4. Cấp session qua Dependency Injection (Depends)

FastAPI dùng Depends để tiêm phụ thuộc. Đây là nơi lý tưởng để cấp một session mỗi request và đảm bảo dọn dẹp dù endpoint thành công hay ném lỗi. Với async, dùng dependency dạng generator:

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

async def get_session() -> AsyncSession:
    async with SessionLocal() as session:
        yield session
    # rời khỏi async-with -> session được đóng, conn trả về pool

# dùng trong endpoint
from typing import Annotated
SessionDep = Annotated[AsyncSession, Depends(get_session)]

async with SessionLocal() bảo đảm session luôn được đóng (trả conn về pool) kể cả khi query ném exception — vì __aexit__ chạy trong mọi trường hợp. Nhờ Annotated, ta khai báo tham số endpoint gọn: async def endpoint(session: SessionDep).

Vòng đời một request:

5. Truy vấn an toàn: LUÔN tham số hoá

Đây là quy tắc bất khả xâm phạm. Không bao giờ nối chuỗi giá trị người dùng vào SQL. Trong ngân hàng, một endpoint tra cứu bị SQL injection có thể lộ toàn bộ số dư khách hàng.

Sai — mở toang cửa cho injection:

# TUYỆT ĐỐI KHÔNG LÀM
q = f"SELECT * FROM transactions WHERE account_id = {account_id}"

Nếu account_id là chuỗi "0 OR 1=1", câu trên trả về toàn bộ bảng.

Đúng — dùng bind parameter, để driver tự escape:

from sqlalchemy import text

stmt = text(
    "SELECT id, account_id, amount, kind, created_at "
    "FROM transactions WHERE account_id = :acc ORDER BY created_at DESC"
)
result = await session.execute(stmt, {"acc": account_id})
rows = result.mappings().all()

Hoặc bằng Core, an toàn tương đương và có type-check tốt hơn:

from sqlalchemy import select, Table, MetaData
# giả định đã reflect/định nghĩa bảng transactions -> tx
stmt = select(tx).where(tx.c.account_id == account_id).order_by(tx.c.created_at.desc())
rows = (await session.execute(stmt)).mappings().all()

Điểm mấu chốt: giá trị đi qua placeholder (:acc) chứ không nhúng vào chuỗi SQL. Driver gửi câu lệnh và giá trị tách biệt, nên dữ liệu không bao giờ bị hiểu là mã.

6. Phân trang: offset vs keyset

Bảng transactions của một ngân hàng dễ đạt hàng chục triệu dòng. Không bao giờ trả tất cả — luôn phân trang.

Offset pagination (LIMIT/OFFSET)

Đơn giản, hỗ trợ "nhảy trang":

SELECT ... FROM transactions WHERE account_id = :acc
ORDER BY created_at DESC
LIMIT :limit OFFSET :offset;

Nhược điểm nghiêm trọng: OFFSET 1000000 buộc Postgres quét và bỏ một triệu dòng trước khi lấy trang cần — càng về sau càng chậm. Ngoài ra nếu có giao dịch mới chèn vào giữa lúc phân trang, các trang bị lệch (thấy trùng hoặc sót dòng).

Keyset (cursor) pagination

Cho dữ liệu lớn và cuộn vô hạn, dùng keyset: nhớ giá trị khoá của dòng cuối trang trước làm điểm neo, lấy tiếp từ đó. Vì có index trên (account_id, created_at), DB "nhảy thẳng" tới vị trí — thời gian gần như không đổi dù ở trang thứ mấy.

Đây chính là truy vấn endpoint dùng trên sandbox — lịch sử giao dịch theo account_id, phân trang keyset theo created_at:

-- ▶ Chạy được
SELECT id, account_id, amount, kind, created_at
FROM transactions
WHERE account_id = 1
  AND created_at < TIMESTAMP '2024-01-01 00:00:00'
ORDER BY created_at DESC
LIMIT 20;

Trang đầu bỏ điều kiện created_at < ... (hoặc dùng mốc now()); các trang sau truyền created_at của dòng cuối vừa trả. Cursor thường được mã hoá base64 để client không cần hiểu cấu trúc bên trong.

7. Map kết quả sang response_model

Endpoint trả JSON qua Pydantic response_model (đã bàn ở bài validation dữ liệu). Row từ SQLAlchemy là mapping, Pydantic v2 map trực tiếp:

from pydantic import BaseModel
from datetime import datetime
from decimal import Decimal

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

class TxPage(BaseModel):
    items: list[TransactionOut]
    next_cursor: str | None

response_model không chỉ tuần tự hoá — nó còn lọc field, nên dù query trả cột thừa, response chỉ chứa field đã khai báo. Đây là lá chắn tránh vô tình lộ cột nhạy cảm. Lưu ý dùng Decimal cho tiền tệ (amount, balance) để không mất chính xác như float.

8. Endpoint hoàn chỉnh và tổng hợp số dư

Ghép mọi thứ lại:

from fastapi import APIRouter, HTTPException, Query
import base64, json
from sqlalchemy import text

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

def encode_cursor(dt: datetime) -> str:
    return base64.urlsafe_b64encode(json.dumps({"ts": dt.isoformat()}).encode()).decode()

@router.get("/{account_id}/transactions", response_model=TxPage)
async def list_transactions(
    account_id: int,
    session: SessionDep,
    cursor: str | None = None,
    limit: int = Query(20, ge=1, le=100),
):
    params = {"acc": account_id, "limit": limit}
    where_cursor = ""
    if cursor:
        ts = json.loads(base64.urlsafe_b64decode(cursor))["ts"]
        where_cursor = "AND created_at < :ts"
        params["ts"] = ts

    stmt = text(
        f"SELECT id, account_id, amount, kind, created_at "
        f"FROM transactions WHERE account_id = :acc {where_cursor} "
        f"ORDER BY created_at DESC LIMIT :limit"
    )
    rows = (await session.execute(stmt, params)).mappings().all()
    items = [TransactionOut(**r) for r in rows]
    next_cursor = encode_cursor(items[-1].created_at) if len(items) == limit else None
    return TxPage(items=items, next_cursor=next_cursor)

Chú ý: f-string ở đây chỉ ghép mảnh cấu trúc (where_cursor) do server kiểm soát hoàn toàn, còn mọi giá trị người dùng (:acc, :ts, :limit) vẫn đi qua bind parameter — không phá vỡ nguyên tắc chống injection.

Endpoint tổng hợp số dư của một khách theo currency cần JOIN accounts với customers. Đây là truy vấn endpoint dùng trên sandbox:

-- ▶ Chạy được
SELECT c.full_name, a.currency,
       COUNT(*) AS num_accounts,
       SUM(a.balance) AS total_balance
FROM accounts a
JOIN customers c ON c.id = a.customer_id
WHERE c.id = 1
GROUP BY c.full_name, a.currency
ORDER BY a.currency;

9. Transaction: khi nào cần

API tra cứu chỉ đọc thường không cần transaction tường minh — mỗi SELECT đã nhất quán. Nhưng khi một request thực hiện nhiều thao tác phải "tất cả hoặc không" (ví dụ ghi log kiểm toán kèm cập nhật trạng thái), phải bọc trong một transaction:

async with session.begin():          # BEGIN
    await session.execute(stmt_1)
    await session.execute(stmt_2)
# rời block không lỗi -> COMMIT tự động; có exception -> ROLLBACK

Nguyên tắc: giữ transaction ngắn nhất có thể. Transaction dài giữ conn và giữ lock, làm nghẽn pool và tăng nguy cơ deadlock.

10. Ánh xạ lỗi DB sang HTTP

Đừng để traceback của SQLAlchemy lọt ra client (lộ cấu trúc DB và gây 500 chung chung). Bắt lỗi và trả mã HTTP đúng ngữ nghĩa:

from sqlalchemy.exc import OperationalError, DBAPIError

@router.get("/{account_id}/summary")
async def account_summary(account_id: int, session: SessionDep):
    try:
        row = (await session.execute(
            text("SELECT id, balance, currency FROM accounts WHERE id = :id"),
            {"id": account_id},
        )).mappings().first()
    except OperationalError:
        raise HTTPException(status_code=503, detail="Database unavailable")
    if row is None:
        raise HTTPException(status_code=404, detail="Account not found")
    return dict(row)

OperationalError (mất kết nối, pool timeout) → 503 để client biết đây là sự cố tạm thời và có thể retry. first() trả None khi không có dòng → 404. Nên đặt thêm một exception handler toàn cục cho DBAPIError chưa lường trước, trả 500 với thông điệp trung tính đồng thời ghi log chi tiết phía server.

11. Tách tầng: repository / service

Nhét SQL thẳng vào endpoint làm code khó test và khó tái dùng. Cấu trúc thực chiến tách ba tầng:

• Router/endpoint: nhận request, validate tham số, gọi service, map ra response_model.
• Service: logic nghiệp vụ (ví dụ quy đổi tỉ giá, quy tắc phân trang).
• Repository: chỉ biết đọc/ghi DB — nhận session, trả dữ liệu thô. Đây là chỗ duy nhất chứa câu SQL.

Lợi ích: đổi từ SQL thô sang ORM chỉ động vào repository; unit test service bằng repository giả (mock) không cần DB thật. Đây là ứng dụng cụ thể của nguyên tắc phân tách trách nhiệm trong bài OOP & clean code.

Use case thực tế

Phòng dịch vụ khách hàng NCB cần màn hình tra cứu: nhân viên nhập ID khách, xem tổng số dư theo loại tiền và cuộn lịch sử giao dịch của từng tài khoản. Bảng transactions khoảng 40 triệu dòng.

Thiết kế và kết quả:

1. Endpoint số dư /customers/{id}/balance-summary dùng truy vấn JOIN ở mục 8 — trả tổng số dư theo currency. Có index accounts(customer_id), chạy dưới 5 ms.
2. Endpoint lịch sử /accounts/{id}/transactions dùng keyset theo created_at (mục 6), index transactions(account_id, created_at DESC), limit=20. Trang đầu và trang thứ 500 đều trả trong ~8 ms — nếu dùng OFFSET 10000 thì trang xa lên tới hàng trăm ms.
3. Pool: 3 instance × pool_size=8, max_overflow=12 = tối đa 60 conn, an toàn dưới max_connections=100. Bật pool_pre_ping sau khi gặp lỗi conn chết lúc 2h sáng (firewall đóng conn nhàn rỗi).
4. Lỗi: khách không tồn tại → 404; lúc bảo trì DB → 503 để frontend hiện "thử lại sau" thay vì màn hình lỗi trắng.
5. Bảo mật: mọi account_id, customer_id, cursor đều qua bind parameter; account read-only api_ro chỉ có quyền SELECT — kể cả có lỗ hổng cũng không ghi/xoá được.

Ghi nhớ

• Tạo một engine/pool ở cấp ứng dụng; đừng mở kết nối mới mỗi request.
• Cấp session qua Depends dạng async with ... yield để luôn trả conn về pool kể cả khi lỗi.
• Chọn pool: tổng (pool_size + max_overflow) × số instance phải nhỏ hơn max_connections; bật pool_pre_ping.
• Luôn tham số hoá (:param/bind), không bao giờ nối chuỗi giá trị vào SQL — chống SQL injection.
• Dữ liệu lớn dùng keyset pagination theo cột có index (created_at); OFFSET chậm dần và dễ lệch trang.
• Dùng response_model để lọc field và ép kiểu; tiền tệ dùng Decimal.
• async + asyncpg cho endpoint I/O-bound tải cao; transaction giữ ngắn nhất có thể.
• Map lỗi: 404 khi không thấy, 409 khi xung đột, 503 khi mất kết nối, 500 cho bug — không để traceback lọt ra client.
• Tách repository/service để dễ test và thay đổi tầng DB độc lập.