Python 8 — Làm việc với SQL, API & Async

import sqlite3

conn = sqlite3.connect("warehouse.db")
cur = conn.cursor()

cur.execute("""
    CREATE TABLE IF NOT EXISTS events (
        id      INTEGER PRIMARY KEY,
        user_id INTEGER,
        action  TEXT,
        ts      TEXT
    )
""")

cur.execute(
    "INSERT INTO events (user_id, action, ts) VALUES (?, ?, ?)",
    (42, "login", "2026-06-30T08:00:00"),
)
conn.commit()  # bắt buộc để lưu thay đổi

cur.execute("SELECT id, action FROM events WHERE user_id = ?", (42,))
for row in cur.fetchall():
    print(row)  # (1, 'login')

conn.close()

# SAI — lỗ hổng SQL injection
action = "login'; DROP TABLE events; --"
cur.execute(f"SELECT * FROM events WHERE action = '{action}'")  # NGUY HIỂM

# ĐÚNG — để driver chèn giá trị an toàn
cur.execute("SELECT * FROM events WHERE action = ?", (action,))

from sqlalchemy import create_engine, text

# Chuỗi kết nối: dialect+driver://user:pass@host:port/dbname
engine = create_engine("postgresql+psycopg2://etl:secret@localhost:5432/dw")
# Ví dụ SQLite dùng ngay dưới đây:
engine = create_engine("sqlite:///warehouse.db", echo=False)

with engine.connect() as conn:
    result = conn.execute(
        text("SELECT action, COUNT(*) AS n FROM events WHERE user_id = :uid GROUP BY action"),
        {"uid": 42},
    )
    for row in result:
        print(row.action, row.n)   # truy cập theo tên cột
    conn.commit()

from datetime import datetime
from sqlalchemy import String, Integer, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Event(Base):
    __tablename__ = "events"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    user_id: Mapped[int] = mapped_column(Integer, index=True)
    action: Mapped[str] = mapped_column(String(50))
    ts: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    def __repr__(self) -> str:
        return f"<Event {self.action} u={self.user_id}>"

Base.metadata.create_all(engine)   # tạo bảng nếu chưa có

with Session(engine) as session:
    # Thêm
    session.add(Event(user_id=42, action="login"))
    session.add_all([
        Event(user_id=42, action="view"),
        Event(user_id=7,  action="login"),
    ])
    session.commit()

    # Truy vấn (API 2.0)
    from sqlalchemy import select
    stmt = select(Event).where(Event.user_id == 42).order_by(Event.id)
    for ev in session.scalars(stmt):
        print(ev)

    # Cập nhật: sửa thuộc tính rồi commit
    ev = session.get(Event, 1)
    ev.action = "login_ok"
    session.commit()

    # Xoá
    session.delete(ev)
    session.commit()

import requests

resp = requests.get(
    "https://api.example.com/v1/users",
    params={"role": "analyst", "active": "true"},
    headers={"Authorization": "Bearer TOKEN", "Accept": "application/json"},
    timeout=10,   # LUÔN đặt timeout
)

# Kiểm tra status code
if resp.status_code == 200:
    data = resp.json()      # parse JSON thành dict/list
    print(data["users"][0])
else:
    print("Lỗi:", resp.status_code, resp.text)

resp = requests.post(
    "https://api.example.com/v1/events",
    json={"user_id": 42, "action": "purchase"},  # json= tự set Content-Type
    headers={"Authorization": "Bearer TOKEN"},
    timeout=10,
)
resp.raise_for_status()   # ném exception nếu 4xx/5xx
created = resp.json()

import requests
from requests.exceptions import RequestException

def fetch_users(page: int) -> dict:
    try:
        resp = requests.get(
            "https://api.example.com/v1/users",
            params={"page": page},
            timeout=10,
        )
        resp.raise_for_status()           # 4xx/5xx -> HTTPError
        return resp.json()
    except requests.Timeout:
        raise RuntimeError(f"Timeout ở trang {page}")
    except requests.HTTPError as e:
        # phân biệt lỗi client vs server
        code = e.response.status_code
        if code == 429:
            raise RuntimeError("Bị rate limit")
        raise RuntimeError(f"HTTP {code}: {e.response.text[:200]}")
    except RequestException as e:
        raise RuntimeError(f"Lỗi mạng: {e}")

import time
import requests

def fetch_all(url: str, token: str) -> list[dict]:
    items, page = [], 1
    while True:
        resp = requests.get(
            url,
            params={"page": page, "per_page": 100},
            headers={"Authorization": f"Bearer {token}"},
            timeout=10,
        )
        if resp.status_code == 429:
            # tôn trọng header Retry-After nếu có
            wait = int(resp.headers.get("Retry-After", "5"))
            time.sleep(wait)
            continue
        resp.raise_for_status()
        batch = resp.json()["data"]
        if not batch:
            break                # hết dữ liệu
        items.extend(batch)
        page += 1
        time.sleep(0.2)          # giãn nhịp, tránh chạm rate limit
    return items

import httpx

# Đồng bộ — dùng Client để tái sử dụng kết nối
with httpx.Client(timeout=10) as client:
    r = client.get("https://api.example.com/v1/users", params={"page": 1})
    r.raise_for_status()
    print(r.json())

import asyncio

async def say(msg: str, delay: float) -> str:
    await asyncio.sleep(delay)   # mô phỏng chờ I/O, không block luồng
    print(msg)
    return msg

async def main():
    # await tuần tự — mất ~3s
    await say("một", 1)
    await say("hai", 2)

asyncio.run(main())   # khởi tạo event loop và chạy coroutine gốc

import asyncio
import httpx

async def fetch(client: httpx.AsyncClient, url: str) -> dict:
    r = await client.get(url, timeout=10)
    r.raise_for_status()
    return r.json()

async def fetch_many(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        tasks = [fetch(client, u) for u in urls]
        # gather chạy tất cả đồng thời, trả kết quả theo đúng thứ tự
        return await asyncio.gather(*tasks)

urls = [f"https://api.example.com/v1/page/{i}" for i in range(1, 21)]
results = asyncio.gather  # minh hoạ
data = asyncio.run(fetch_many(urls))
print(f"Lấy {len(data)} trang")

async def fetch_limited(sem, client, url):
    async with sem:                      # tối đa N request cùng lúc
        return await fetch(client, url)

async def fetch_throttled(urls, concurrency=5):
    sem = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient() as client:
        tasks = [fetch_limited(sem, client, u) for u in urls]
        return await asyncio.gather(*tasks)

# file: test_pipeline.py

def parse_event(raw: dict) -> dict:
    return {"user": raw["user_id"], "action": raw["action"].lower()}

def test_parse_event():
    out = parse_event({"user_id": 42, "action": "LOGIN"})
    assert out == {"user": 42, "action": "login"}

def test_parse_event_missing_key():
    import pytest
    with pytest.raises(KeyError):
        parse_event({"user_id": 42})

import pytest

# Tham số hoá: một test, nhiều bộ dữ liệu
@pytest.mark.parametrize("inp,expected", [
    ("LOGIN", "login"),
    ("View",  "view"),
])
def test_normalize(inp, expected):
    assert inp.lower() == expected

# Fixture: chuẩn bị tài nguyên dùng chung
@pytest.fixture
def sample_rows():
    return [{"user_id": 1, "action": "X"}, {"user_id": 2, "action": "Y"}]

def test_count(sample_rows):
    assert len(sample_rows) == 2