Python DE 6 — Lấy dữ liệu từ API & ingestion

Mở đầu: dữ liệu không nằm sẵn trong database

Rất nhiều dữ liệu quan trọng nhất của một ngân hàng lại không nằm trong database của bạn. Danh sách giao dịch thẻ nằm ở hệ thống của đối tác thanh toán. Tỷ giá ngoại tệ đến từ một nhà cung cấp dữ liệu thị trường. Điểm tín dụng, thông tin định danh, danh sách đen — tất cả đều là những hệ thống bên ngoài, và cách duy nhất để lấy được chúng là gọi API.

Với data engineer, "ingestion qua API" là công việc hằng ngày: kéo dữ liệu từ một endpoint HTTP về, làm sạch, rồi nạp vào kho. Nghe đơn giản — "gọi một cái URL rồi lưu kết quả" — nhưng thực tế đầy cạm bẫy: token hết hạn giữa chừng, API trả 429 vì bạn gọi quá nhanh, dữ liệu chia làm 4.000 trang, mạng chập chờn làm request treo, và schema của đối tác âm thầm đổi.

Bài này đi sâu vào cách thu thập dữ liệu từ REST API một cách bền vững (production-grade): chọn thư viện, xử lý xác thực và phân trang, chịu đựng rate limit bằng retry có backoff, mở rộng quy mô bằng async, và các mẫu ingestion kinh điển — full load và incremental với high-water mark. Ta giả định bạn đã đọc Python DE 4 — File & I/O (lưu Parquet) và Python DE 5 — Kết nối database (lưu state và nạp kết quả).

Quy ước: sandbox của series là PostgreSQL, Python không chạy trực tiếp ở đây. Các khối python chỉ để minh hoạ mẫu code, không phải để chạy trong sandbox.

Nguồn dữ liệu qua API và vì sao REST/JSON thống trị

Trong DE ngân hàng, API xuất hiện dưới ba nhóm nguồn chính:

• SaaS & dịch vụ nội bộ: CRM, cổng thanh toán, hệ thống thẻ, hệ thống loyalty. Thường là REST + JSON, xác thực bằng OAuth2 hoặc API key.
• Ngân hàng lõi (core banking) & đối tác: API cung cấp giao dịch, số dư, hồ sơ khoản vay. Thường khắt khe hơn về bảo mật — HMAC ký request, IP whitelist, mTLS.
• Dữ liệu công khai / thị trường: tỷ giá, lãi suất tham chiếu, dữ liệu vĩ mô. Thường có rate limit chặt và phân trang lớn.

Vì sao REST/JSON phổ biến nhất? Vì nó chạy trên HTTP — thứ mà mọi tường lửa, load balancer, proxy đều hiểu — và JSON thì con người đọc được, mọi ngôn ngữ đều parse được. Có SOAP/XML (còn ở các hệ thống cũ), gRPC (nội bộ, hiệu năng cao), GraphQL (client tự chọn field), nhưng với ingestion từ bên ngoài, REST/JSON vẫn là mẫu số chung. Cả bài này ta tập trung vào REST/JSON.

Thư viện: requests vs httpx

Có hai lựa chọn phổ biến để gọi HTTP trong Python:

requests — thư viện HTTP kinh điển, cực kỳ phổ biến, API trực giác. Nhược điểm: chỉ đồng bộ (blocking), không hỗ trợ HTTP/2, và dự án gần như không còn phát triển tính năng mới.

httpx — thư viện hiện đại, API gần như tương thích requests nhưng bổ sung: hỗ trợ cả đồng bộ lẫn async trong cùng một thư viện, HTTP/2 (khi bật), timeout mặc định hợp lý. Đây là lựa chọn khuyến nghị cho code mới, đặc biệt khi bạn cần async để mở rộng quy mô.

Điểm chung quan trọng nhất cho DE: luôn tái dùng session/client. Mỗi request mới nếu tạo kết nối TCP mới sẽ tốn thời gian bắt tay (và bắt tay TLS). Một Session (requests) hay Client (httpx) giữ connection pool, tái dùng kết nối đã mở — nhanh hơn nhiều khi gọi hàng nghìn request tới cùng một host.

# (minh hoạ) — httpx.Client tái dùng connection, timeout rõ ràng
import httpx

# Một client dùng chung cho cả pipeline, đặt timeout để không treo mãi mãi
client = httpx.Client(
    base_url="https://api.doitac.vn/v1",
    timeout=httpx.Timeout(10.0, connect=5.0),   # 10s tổng, 5s cho lúc connect
    headers={"Accept": "application/json"},
    http2=True,                                  # bật HTTP/2 nếu upstream hỗ trợ
    limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
)

resp = client.get("/transactions", params={"date": "2026-06-30"})
resp.raise_for_status()   # ném lỗi nếu 4xx/5xx thay vì im lặng
data = resp.json()

raise_for_status() là thói quen bắt buộc: nếu không gọi, một response 500 vẫn "thành công" về mặt code và bạn sẽ cố parse JSON của trang lỗi.

Xác thực: đừng bao giờ hardcode

Gần như mọi API nghiêm túc đều yêu cầu xác thực. Ba kiểu thường gặp:

• API key: một chuỗi bí mật, gửi qua header (X-API-Key) hoặc query param. Đơn giản nhưng không có cơ chế hết hạn tự động.
• Bearer / OAuth2 token: bạn gọi endpoint token với client_id/client_secret, nhận về access_token có thời hạn (ví dụ 1 giờ), rồi đính kèm Authorization: Bearer <token> vào mỗi request. Khi token hết hạn (nhận 401), bạn refresh rồi thử lại.
• HMAC: bạn ký nội dung request bằng một secret dùng chung; server ký lại và so sánh. Chống giả mạo, thường thấy ở API ngân hàng/đối tác.

Nguyên tắc bất di bất dịch: không hardcode secret vào code hay commit lên git. Đọc từ biến môi trường hoặc một secret manager (Vault, AWS Secrets Manager, GCP Secret Manager).

# (minh hoạ) — OAuth2 client-credentials với tự refresh token
import os, time, httpx

class TokenManager:
    def __init__(self, client: httpx.Client):
        self._client = client
        self._token = None
        self._expires_at = 0.0

    def get(self) -> str:
        # Refresh trước 60s khi token sắp hết hạn (tránh dùng token "gần chết")
        if self._token is None or time.time() > self._expires_at - 60:
            r = self._client.post("/oauth/token", data={
                "grant_type": "client_credentials",
                "client_id": os.environ["API_CLIENT_ID"],       # từ env, không hardcode
                "client_secret": os.environ["API_CLIENT_SECRET"],
            })
            r.raise_for_status()
            payload = r.json()
            self._token = payload["access_token"]
            self._expires_at = time.time() + payload["expires_in"]
        return self._token

Phân trang: lấy hết mọi trang mà không nổ RAM

Hiếm có API nào trả toàn bộ dữ liệu trong một response. Chúng chia thành trang (page), và bạn phải lặp cho tới khi hết. Ba kiểu phân trang chính:

• offset/limit: ?offset=0&limit=100, ?offset=100&limit=100... Đơn giản nhưng dễ trùng/sót nếu dữ liệu thay đổi giữa các trang.
• cursor: mỗi response trả về một next_cursor (con trỏ mờ đục); bạn gửi lại nó ở request kế. Ổn định hơn khi dữ liệu đang biến động.
• page token: giống cursor nhưng gọi là next_page_token (kiểu Google API).

Mẹo DE cốt lõi: dùng generator (yield) để trả từng bản ghi hoặc từng trang. Như vậy bạn không giữ toàn bộ kết quả trong RAM — có thể xử lý một triệu bản ghi mà bộ nhớ vẫn phẳng, vì mỗi trang được tiêu thụ rồi giải phóng.

# (minh hoạ) — generator lặp hết trang theo cursor
from typing import Iterator

def iter_transactions(client: httpx.Client, since: str) -> Iterator[dict]:
    cursor = None
    while True:
        params = {"since": since, "limit": 500}
        if cursor:
            params["cursor"] = cursor
        r = client.get("/transactions", params=params)
        r.raise_for_status()
        body = r.json()

        for record in body["data"]:   # trả từng bản ghi, không gom hết vào list
            yield record

        cursor = body.get("next_cursor")
        if not cursor:                # hết trang -> dừng
            break

# Người dùng chỉ việc: for tx in iter_transactions(client, "2026-06-30"): ...

Generator biến "lấy hết mọi trang" thành một luồng dữ liệu lười (lazy stream) — mẫu này gắn rất tự nhiên với việc ghi Parquet theo lô (batch) mà không cần vật chất hoá toàn bộ.

Rate limit & độ bền: chịu đựng thế giới thực

Mạng sẽ hỏng. API sẽ giới hạn bạn. Đây là phần phân biệt một script "chạy được trên máy tôi" với một pipeline production.

Tôn trọng 429 và Retry-After

Khi bạn gọi quá nhanh, API trả về HTTP 429 Too Many Requests, thường kèm header Retry-After (số giây cần đợi). Hành xử đúng đắn: đọc Retry-After và ngủ đúng bấy nhiêu, đừng đoán.

Exponential backoff + jitter

Với lỗi tạm thời (429, 500, 502, 503, timeout mạng), ta thử lại — nhưng không thử lại ngay lập tức và không thử lại mãi. Mẫu chuẩn công nghiệp là exponential backoff: đợi 1s, rồi 2s, 4s, 8s... (nhân đôi mỗi lần). Quan trọng không kém là thêm jitter — một lượng ngẫu nhiên vào thời gian đợi. Không có jitter, khi một API sập rồi hồi phục, tất cả client đồng loạt retry cùng một thời điểm và lại đánh sập nó (hiệu ứng "thundering herd"). Jitter làm lệch nhịp các client ra. Với nhiều client đồng thời, full jitter (đợi một khoảng ngẫu nhiên từ 0 đến giá trị backoff hiện tại) là mặc định được khuyến nghị.

tenacity: retry khai báo

Thay vì tự viết vòng lặp retry, dùng thư viện tenacity — cho phép khai báo chính sách retry (số lần thử, kiểu chờ, điều kiện retry) một cách gọn gàng, hỗ trợ cả hàm đồng bộ lẫn async.

# (minh hoạ) — retry chỉ với lỗi tạm thời, backoff mũ + jitter
import httpx
from tenacity import (
    retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception,
)

def is_retryable(exc: Exception) -> bool:
    # Retry với lỗi mạng, và với 429/5xx; KHÔNG retry với 4xx còn lại
    if isinstance(exc, (httpx.TimeoutException, httpx.TransportError)):
        return True
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code in (429, 500, 502, 503, 504)
    return False

@retry(
    stop=stop_after_attempt(5),                       # tối đa 5 lần
    wait=wait_exponential_jitter(initial=1, max=60),  # 1,2,4,8... + jitter, trần 60s
    retry=retry_if_exception(is_retryable),
    reraise=True,                                     # hết lượt thì ném lỗi thật
)
def fetch(client: httpx.Client, url: str, **kw) -> httpx.Response:
    r = client.get(url, **kw)
    r.raise_for_status()
    return r

Timeout, circuit breaker, idempotency

• Timeout: luôn đặt timeout (như ở httpx.Timeout phía trên). Một request không timeout có thể treo pipeline vô hạn.
• Circuit breaker: nếu một API liên tục lỗi, ngừng gọi nó một lúc (mở "cầu dao") thay vì tiếp tục dội request. Tránh lãng phí và cho hệ thống kia thời gian hồi phục.
• Idempotency: retry chỉ an toàn khi thao tác lặp lại nhiều lần cho cùng kết quả. Với GET thì luôn an toàn. Với POST tạo dữ liệu, nhiều API hỗ trợ header Idempotency-Key — gửi cùng một key để server không tạo bản ghi trùng khi bạn retry.

Async cho quy mô: hàng nghìn request

Khi phải gọi hàng nghìn endpoint (ví dụ: lấy chi tiết cho 5.000 khách hàng, mỗi người một request), gọi tuần tự sẽ chậm khủng khiếp — vì phần lớn thời gian là chờ mạng, không phải tính toán. Đây chính là lúc async toả sáng.

Với asyncio + httpx.AsyncClient, trong khi một request đang chờ mạng, chương trình chuyển sang gửi/xử lý request khác. Điểm mấu chốt: giới hạn số request đồng thời bằng một Semaphore, nếu không bạn sẽ tự bắn 5.000 request cùng lúc và nhận về một rừng 429 (hoặc bị block IP).

# (minh hoạ) — async fan-out có kiểm soát bằng semaphore
import asyncio, httpx

async def fetch_one(client, sem, cust_id):
    async with sem:                     # tối đa N request "trong bay" cùng lúc
        r = await client.get(f"/customers/{cust_id}")
        r.raise_for_status()
        return r.json()

async def fetch_all(cust_ids: list[str]):
    limits = httpx.Limits(max_connections=50)
    sem = asyncio.Semaphore(20)         # giữ đồng thời ở mức API chịu được
    async with httpx.AsyncClient(base_url="https://api.doitac.vn/v1",
                                 limits=limits, http2=True, timeout=10) as client:
        tasks = [fetch_one(client, sem, cid) for cid in cust_ids]
        return await asyncio.gather(*tasks, return_exceptions=True)

# results có thể chứa cả dict lẫn Exception -> tách lỗi ra để xử lý riêng

Khi nào async đáng công? Khi khối lượng công việc là I/O-bound (nhiều request chờ mạng), số lượng lớn, và cùng chạm một vài host. Khi công việc là CPU-bound (biến đổi dữ liệu nặng), async không giúp gì — đó là chuyện của tiến trình/đa luồng, sẽ bàn kỹ ở Python DE 7 — Concurrency & hiệu năng. Đừng dùng async chỉ vì nó "hiện đại"; với vài chục request, code đồng bộ đơn giản hơn và đủ nhanh.

Webhook & push: khi API chủ động gọi bạn

Mọi thứ ở trên là mô hình pull — bạn chủ động đi lấy. Nhiều hệ thống hiện đại dùng mô hình push qua webhook: khi có sự kiện (giao dịch mới, thanh toán thành công), nhà cung cấp gửi một HTTP POST tới URL của bạn. Bạn không phải poll liên tục; dữ liệu tới ngay khi có.

Ba điểm sống còn khi nhận webhook:

1. Xác thực chữ ký: bất kỳ ai biết URL đều có thể POST giả. Nhà cung cấp thường ký payload bằng HMAC (một secret dùng chung) và gửi chữ ký trong header. Bạn phải tính lại HMAC trên body nhận được và so sánh an toàn (hmac.compare_digest) trước khi tin.
2. Trả 200 nhanh, xử lý sau: nhà cung cấp thường timeout sau vài giây và sẽ retry nếu không nhận được 200. Vì vậy đừng xử lý nặng ngay trong handler — hãy đẩy sự kiện vào queue (Kafka, SQS, Redis) rồi trả 200 ngay. Một worker riêng sẽ tiêu thụ queue sau.
3. Chống trùng (idempotency): webhook có thể được gửi nhiều lần cho cùng một sự kiện. Dùng event_id để dedup.

# (minh hoạ) — verify chữ ký HMAC của webhook trước khi tin
import hmac, hashlib

def verify(secret: bytes, raw_body: bytes, signature_header: str) -> bool:
    expected = hmac.new(secret, raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature_header)  # so sánh chống timing-attack

Mẫu ingestion: full load vs incremental

Đây là trái tim của thiết kế pipeline ingestion.

Full load — mỗi lần chạy, lấy toàn bộ dữ liệu và ghi đè. Đơn giản, không sợ sót; nhưng tốn kém và không khả thi khi dữ liệu lớn hoặc API giới hạn khối lượng. Phù hợp cho bảng nhỏ, tương đối tĩnh (danh mục chi nhánh, mã sản phẩm).

Incremental load — mỗi lần chỉ lấy phần thay đổi từ lần chạy trước. Đây là mẫu chủ lực. Bạn lưu một high-water mark (mốc nước cao) — thường là timestamp lớn nhất hoặc cursor cuối cùng đã xử lý — vào một chỗ bền vững (bảng state trong PostgreSQL). Lần chạy sau, bạn hỏi API "cho tôi mọi thứ sau mốc này".

Vài chi tiết quyết định độ chắc chắn:

• Dedup: dữ liệu ở biên (cùng timestamp với mark) dễ bị lấy trùng. Thường dùng >= mark rồi khử trùng theo khoá nghiệp vụ (transaction_id). Chấp nhận lấy dư một chút còn hơn sót.
• Schema drift: đối tác thêm/xoá/đổi field mà không báo. Lưu raw trước (xem dưới) giúp bạn không mất dữ liệu ngay cả khi bước transform vỡ, và validate bằng pydantic giúp phát hiện sớm.
• Lưu raw (landing zone) trước khi transform: nguyên tắc vàng. Ghi payload thô (JSON hoặc Parquet) xuống một vùng bất biến trước, rồi mới transform từ đó. Nếu logic transform sai, bạn chạy lại từ raw mà không phải gọi lại API. Đây chính là tầng Bronze/landing trong kiến trúc medallion của lakehouse.
• Lưu Parquet: định dạng cột nén tốt, đọc nhanh, giữ schema — lý tưởng cho landing (xem Python DE 4 — File & I/O). Ghi theo partition (ví dụ theo ngày) để dễ chạy lại từng phần.

# (minh hoạ) — incremental theo cursor, lưu state + ghi Parquet
import pyarrow as pa, pyarrow.parquet as pq
from sqlalchemy import text

def load_state(conn, source: str) -> str | None:
    row = conn.execute(text(
        "SELECT hwm FROM ingest_state WHERE source = :s"), {"s": source}).first()
    return row[0] if row else None

def save_state(conn, source: str, hwm: str) -> None:
    conn.execute(text("""
        INSERT INTO ingest_state (source, hwm) VALUES (:s, :h)
        ON CONFLICT (source) DO UPDATE SET hwm = EXCLUDED.hwm
    """), {"s": source, "h": hwm})

def run_incremental(client, conn, source="partner_tx"):
    hwm = load_state(conn, source) or "1970-01-01T00:00:00Z"
    batch, max_ts = [], hwm
    for tx in iter_transactions(client, since=hwm):   # generator phân trang ở trên
        batch.append(tx)
        max_ts = max(max_ts, tx["created_at"])        # theo dõi mốc mới nhất
    if batch:
        table = pa.Table.from_pylist(batch)
        pq.write_table(table, f"landing/{source}/dt={max_ts[:10]}.parquet")
        save_state(conn, source, max_ts)              # chỉ lưu HWM SAU khi ghi thành công

Thứ tự ở dòng cuối rất quan trọng: ghi dữ liệu thành công rồi mới cập nhật high-water mark. Nếu ghi state trước mà ghi Parquet lỗi, bạn sẽ vĩnh viễn bỏ sót lô đó ở lần sau.

Chất lượng & validate: tin nhưng phải kiểm

API bên ngoài là dữ liệu bạn không kiểm soát. Một field bỗng thành null, một số tiền về dưới dạng chuỗi, một bản ghi thiếu khoá — nếu không kiểm, rác sẽ trôi thẳng vào kho.

• pydantic validate payload: định nghĩa model mô tả bản ghi nên trông thế nào; pydantic ép kiểu và báo lỗi khi sai. Bắt schema drift ngay tại cổng vào.
• Xử lý lỗi từng bản ghi: một bản ghi hỏng không được làm chết cả lô. Bọc validate trong try/except mỗi bản ghi, đẩy bản lỗi vào "dead-letter" (log/bảng riêng) rồi tiếp tục.
• Logging: log số bản ghi lấy được, số bản lỗi, HWM cũ/mới mỗi lần chạy. Đây là bằng chứng để điều tra khi số liệu bất thường.

# (minh hoạ) — validate từng bản ghi, không để một bản hỏng giết cả lô
from pydantic import BaseModel, ValidationError
from datetime import datetime

class Transaction(BaseModel):
    transaction_id: str
    amount: float          # pydantic ép "123.45" -> 123.45
    currency: str
    created_at: datetime

good, bad = [], []
for raw in iter_transactions(client, since=hwm):
    try:
        good.append(Transaction.model_validate(raw))
    except ValidationError as e:
        bad.append({"raw": raw, "error": str(e)})   # dead-letter, xử lý sau

Cuối cùng, ingestion không sống một mình — nó là một bước trong pipeline được điều phối (orchestrated). Trong thực tế, script này chạy dưới dạng một task Airflow (hoặc Dagster/Prefect), có lịch, có retry ở tầng orchestrator, có cảnh báo. Cách gọi API/DB trong Airflow qua hook/operator được bàn ở Airflow — Operators & Hooks.

Use case thực tế: nạp incremental giao dịch từ API đối tác

Bối cảnh: đối tác thanh toán cung cấp API /transactions trả giao dịch thẻ, phân trang bằng cursor, xác thực OAuth2, rate limit 100 request/phút. Ta cần nạp incremental mỗi giờ vào lakehouse.

Thiết kế theo đúng các mảnh ghép ở trên:

1. Client dùng chung (httpx.Client, HTTP/2, timeout 10s, connection pool) + TokenManager tự refresh OAuth2.
2. Đọc state: lấy high-water mark (timestamp giao dịch mới nhất đã nạp) từ bảng ingest_state trong PostgreSQL.
3. Lấy dữ liệu qua generator phân trang cursor, chỉ những giao dịch created_at >= hwm (lấy >= để không sót ở biên, chấp nhận dedup).
4. Độ bền: mỗi lần gọi trang bọc trong tenacity — retry với backoff mũ + jitter cho 429/5xx, tôn trọng Retry-After.
5. Validate từng bản ghi bằng pydantic; bản hỏng đẩy vào dead-letter, không làm chết lô.
6. Ghi RAW ra Parquet partition theo ngày (landing/partner_tx/dt=2026-06-30.parquet) — tầng landing bất biến.
7. Dedup + cập nhật state: khử trùng theo transaction_id, rồi ghi HWM mới vào ingest_state chỉ sau khi Parquet đã ghi xong.
8. Điều phối: toàn bộ là một task Airflow chạy mỗi giờ, retry 3 lần ở tầng orchestrator, gửi cảnh báo Slack nếu số bản lỗi vượt ngưỡng.

Kết quả: nếu API sập lúc 2 giờ sáng, task fail, HWM không đổi; lúc 3 giờ nó chạy lại từ đúng mốc cũ và không sót giao dịch nào. Nếu logic transform xuống Silver sai, ta chạy lại từ Parquet landing mà không cần gọi lại API. Đó là ingestion bền vững.

Ghi nhớ

• Dùng httpx, tái dùng client: connection pool + timeout rõ ràng là mặc định, không phải tuỳ chọn. httpx cho bạn cả đồng bộ lẫn async trong một thư viện.
• Không bao giờ hardcode secret: env hoặc secret manager. Token OAuth2 phải tự refresh trước khi hết hạn.
• Phân trang bằng generator: yield từng bản ghi/trang để không giữ hết trong RAM; lặp tới khi hết cursor/token.
• Chịu đựng thế giới thực: tôn trọng 429/Retry-After, exponential backoff + jitter (dùng tenacity), luôn đặt timeout, chỉ retry lỗi tạm thời, đảm bảo idempotency.
• Async chỉ khi I/O-bound & quy mô lớn: dùng AsyncClient + Semaphore để giới hạn đồng thời — nếu không sẽ tự gây 429. CPU-bound thì async vô ích (bài 7).
• Incremental > full load cho dữ liệu lớn: lưu high-water mark, cập nhật state chỉ sau khi ghi dữ liệu thành công.
• Lưu raw trước, transform sau: landing Parquet bất biến giúp chạy lại mà không gọi lại API; validate bằng pydantic bắt schema drift; lỗi từng bản ghi đẩy vào dead-letter.
• Ingestion sống trong orchestrator: đóng gói thành task Airflow để có lịch, retry, và cảnh báo.

Bài tiếp theo: Python DE 7 — Concurrency & hiệu năng — đi sâu asyncio, threading, multiprocessing và khi nào dùng cái nào. Xem lại: Python DE 4 — File & I/O, Python DE 5 — Kết nối database.