1 bài viết được gắn thẻ "Async"

from typing import Annotated, Literal, Protocol
from collections.abc import Sequence, AsyncIterator
from decimal import Decimal


def tinh_tong_gia(danh_sach: Sequence[dict]) -> Decimal:
    return sum((Decimal(str(item["gia"])) for item in danh_sach), Decimal(0))


TrangThai = Literal["cho_xu_ly", "dang_giao", "hoan_thanh", "da_huy"]


def cap_nhat_trang_thai(don_hang_id: int, trang_thai: TrangThai) -> dict:
    ...


class CoTheLuu(Protocol):
    """Protocol — duck typing có kiểm tra kiểu."""

    def luu(self) -> None: ...

import asyncio
import time
import httpx


# ❌ Chặn event loop — toàn bộ server đứng khi hàm này chạy
async def goi_api_sai(url: str) -> dict:
    response = httpx.get(url)          # httpx.get là SYNC!
    return response.json()


# ✅ Nhường quyền điều khiển trong lúc chờ
async def goi_api_dung(url: str) -> dict:
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.get(url)
        return response.json()


# ✅ Chạy nhiều request đồng thời thay vì tuần tự
async def goi_nhieu_url(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=10) as client:
        return list(await asyncio.gather(*(goi_mot(client, u) for u in urls)))


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

@app.get("/tinh-toan-nang")
def tinh_toan(n: int):                     # ← def (sync): FastAPI chạy trong threadpool
    return {"ket_qua": sum(i * i for i in range(n))}


@app.get("/goi-api-ngoai")
async def goi_api():                       # ← async def: dành cho I/O
    async with httpx.AsyncClient() as c:
        return (await c.get("https://api.example.com")).json()

from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def vong_doi(app: FastAPI):
    """Chạy 1 lần khi server khởi động và 1 lần khi tắt."""
    app.state.http = httpx.AsyncClient(timeout=10)
    print("🚀 Khởi động: đã tạo HTTP client")
    try:
        yield
    finally:
        await app.state.http.aclose()
        print("🛑 Tắt: đã đóng HTTP client")


app = FastAPI(lifespan=vong_doi)

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

app = FastAPI(title="API Ghi Chú", version="1.0.0")

GHI_CHU: dict[int, dict] = {}
_bo_dem = 0


@app.get("/")
async def goc():
    return {"thong_diep": "API đang chạy", "phien_ban": "1.0.0"}


@app.get("/ghi-chu")
async def danh_sach(
    da_xong: bool | None = None,
    tu_khoa: Annotated[str | None, Query(max_length=100)] = None,
    gioi_han: Annotated[int, Query(ge=1, le=100)] = 20,
    offset: Annotated[int, Query(ge=0)] = 0,
):
    ket_qua = list(GHI_CHU.values())

    if da_xong is not None:
        ket_qua = [g for g in ket_qua if g["da_xong"] == da_xong]
    if tu_khoa:
        ket_qua = [g for g in ket_qua if tu_khoa.lower() in g["tieu_de"].lower()]

    return {"tong": len(ket_qua), "du_lieu": ket_qua[offset : offset + gioi_han]}


@app.get("/ghi-chu/{ghi_chu_id}")
async def chi_tiet(ghi_chu_id: Annotated[int, Path(ge=1)]):
    if ghi_chu_id not in GHI_CHU:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Không tìm thấy ghi chú id={ghi_chu_id}",
        )
    return GHI_CHU[ghi_chu_id]

from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, Field, EmailStr, field_validator, model_validator, ConfigDict


class GhiChuVao(BaseModel):
    """Dữ liệu client gửi lên."""

    model_config = ConfigDict(str_strip_whitespace=True)

    tieu_de: str = Field(min_length=1, max_length=200, examples=["Mua sữa"])
    noi_dung: str = Field(default="", max_length=5000)
    do_uu_tien: int = Field(default=2, ge=1, le=5)
    nhan: list[str] = Field(default_factory=list, max_length=10)

    @field_validator("nhan")
    @classmethod
    def chuan_hoa_nhan(cls, v: list[str]) -> list[str]:
        return [n.strip().lower() for n in v if n.strip()]

    @model_validator(mode="after")
    def kiem_tra_logic(self):
        if self.do_uu_tien == 5 and not self.noi_dung:
            raise ValueError("Ghi chú ưu tiên 5 phải có nội dung")
        return self


class GhiChuRa(BaseModel):
    """Dữ liệu trả về — có thêm id và thời gian."""

    model_config = ConfigDict(from_attributes=True)

    id: int
    tieu_de: str
    noi_dung: str
    do_uu_tien: int
    nhan: list[str]
    da_xong: bool
    ngay_tao: datetime


class PhanHoiDanhSach(BaseModel):
    tong: int
    du_lieu: list[GhiChuRa]


@app.post("/ghi-chu", response_model=GhiChuRa, status_code=status.HTTP_201_CREATED)
async def tao_moi(ghi_chu: GhiChuVao) -> GhiChuRa:
    global _bo_dem
    _bo_dem += 1
    ban_ghi = {
        "id": _bo_dem,
        **ghi_chu.model_dump(),
        "da_xong": False,
        "ngay_tao": datetime.now(),
    }
    GHI_CHU[_bo_dem] = ban_ghi
    return GhiChuRa(**ban_ghi)

from pydantic_settings import BaseSettings, SettingsConfigDict


class CauHinh(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    database_url: str
    secret_key: str
    debug: bool = False
    cors_origins: list[str] = ["http://localhost:3000"]
    jwt_thuat_toan: str = "HS256"
    jwt_het_han_phut: int = 60


cau_hinh = CauHinh()      # tự đọc từ .env và biến môi trường

# db.py
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from cau_hinh import cau_hinh

engine = create_async_engine(
    cau_hinh.database_url,
    echo=cau_hinh.debug,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
)

SessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


async def lay_session() -> AsyncIterator[AsyncSession]:
    """Dependency cung cấp session cho mỗi request."""
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

from datetime import datetime
from sqlalchemy import String, Text, Boolean, Integer, DateTime, Index, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.dialects.postgresql import ARRAY

from db import Base


class GhiChu(Base):
    __tablename__ = "ghi_chu"

    id: Mapped[int] = mapped_column(primary_key=True)
    tieu_de: Mapped[str] = mapped_column(String(200), nullable=False)
    noi_dung: Mapped[str] = mapped_column(Text, default="")
    do_uu_tien: Mapped[int] = mapped_column(Integer, default=2)
    nhan: Mapped[list[str]] = mapped_column(ARRAY(String), default=list)
    da_xong: Mapped[bool] = mapped_column(Boolean, default=False)
    chu_so_huu_id: Mapped[int] = mapped_column(ForeignKey("nguoi_dung.id", ondelete="CASCADE"))
    ngay_tao: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

    chu_so_huu: Mapped["NguoiDung"] = relationship(back_populates="ghi_chu")

    __table_args__ = (
        Index("idx_ghi_chu_chu_so_huu_da_xong", "chu_so_huu_id", "da_xong"),
    )

from sqlalchemy import select, func
from sqlalchemy.orm import selectinload


async def lay_danh_sach(session: AsyncSession, user_id: int,
                        da_xong: bool | None = None, gioi_han: int = 20):
    stmt = select(GhiChu).where(GhiChu.chu_so_huu_id == user_id)

    if da_xong is not None:
        stmt = stmt.where(GhiChu.da_xong.is_(da_xong))

    stmt = stmt.order_by(GhiChu.do_uu_tien.desc(), GhiChu.ngay_tao.desc()).limit(gioi_han)

    ket_qua = await session.scalars(stmt)
    return list(ket_qua)


async def dem_theo_uu_tien(session: AsyncSession, user_id: int) -> dict[int, int]:
    stmt = (
        select(GhiChu.do_uu_tien, func.count(GhiChu.id))
        .where(GhiChu.chu_so_huu_id == user_id)
        .group_by(GhiChu.do_uu_tien)
    )
    ket_qua = await session.execute(stmt)
    return {muc: so for muc, so in ket_qua.all()}

# dependencies.py
from typing import Annotated
from fastapi import Depends, HTTPException, Header, status
from sqlalchemy.ext.asyncio import AsyncSession

from db import lay_session
from core.security import giai_ma_token

SessionDep = Annotated[AsyncSession, Depends(lay_session)]


async def nguoi_dung_hien_tai(
    session: SessionDep,
    authorization: Annotated[str | None, Header()] = None,
) -> NguoiDung:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Thiếu token xác thực")

    try:
        payload = giai_ma_token(authorization[7:])
    except TokenHetHan:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Token đã hết hạn")
    except TokenKhongHopLe:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Token không hợp lệ")

    user = await session.get(NguoiDung, int(payload["sub"]))
    if user is None or not user.dang_hoat_dong:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Tài khoản không tồn tại")
    return user


CurrentUser = Annotated[NguoiDung, Depends(nguoi_dung_hien_tai)]


async def yeu_cau_admin(user: CurrentUser) -> NguoiDung:
    if user.vai_tro != "admin":
        raise HTTPException(status.HTTP_403_FORBIDDEN, "Cần quyền quản trị")
    return user


AdminUser = Annotated[NguoiDung, Depends(yeu_cau_admin)]

# routers/ghi_chu.py
from fastapi import APIRouter, status
from dependencies import SessionDep, CurrentUser

router = APIRouter(prefix="/ghi-chu", tags=["Ghi chú"])


@router.get("", response_model=PhanHoiDanhSach)
async def danh_sach(
    session: SessionDep,
    user: CurrentUser,
    da_xong: bool | None = None,
    gioi_han: int = 20,
):
    ds = await lay_danh_sach(session, user.id, da_xong, gioi_han)
    return {"tong": len(ds), "du_lieu": ds}


@router.post("", response_model=GhiChuRa, status_code=status.HTTP_201_CREATED)
async def tao_moi(du_lieu: GhiChuVao, session: SessionDep, user: CurrentUser):
    gc = GhiChu(**du_lieu.model_dump(), chu_so_huu_id=user.id)
    session.add(gc)
    await session.flush()
    await session.refresh(gc)
    return gc

# ❌ Sai — vẫn lộ thông tin "bản ghi tồn tại"
gc = await session.get(GhiChu, ghi_chu_id)
if gc.chu_so_huu_id != user.id:
    raise HTTPException(403)

# ✅ Đúng — không tìm thấy gì cả
stmt = select(GhiChu).where(GhiChu.id == ghi_chu_id, GhiChu.chu_so_huu_id == user.id)
gc = await session.scalar(stmt)
if gc is None:
    raise HTTPException(404, "Không tìm thấy ghi chú")

# core/security.py
from datetime import datetime, timedelta, timezone
import jwt
from passlib.context import CryptContext

from cau_hinh import cau_hinh


class TokenHetHan(Exception): ...
class TokenKhongHopLe(Exception): ...


pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")


def bam_mat_khau(mat_khau: str) -> str:
    return pwd.hash(mat_khau)


def kiem_tra_mat_khau(mat_khau: str, da_bam: str) -> bool:
    return pwd.verify(mat_khau, da_bam)


def tao_token(user_id: int, vai_tro: str = "user") -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": str(user_id),
        "vai_tro": vai_tro,
        "iat": now,
        "exp": now + timedelta(minutes=cau_hinh.jwt_het_han_phut),
    }
    return jwt.encode(payload, cau_hinh.secret_key, algorithm=cau_hinh.jwt_thuat_toan)


def giai_ma_token(token: str) -> dict:
    try:
        return jwt.decode(token, cau_hinh.secret_key, algorithms=[cau_hinh.jwt_thuat_toan])
    except jwt.ExpiredSignatureError as e:
        raise TokenHetHan from e
    except jwt.InvalidTokenError as e:
        raise TokenKhongHopLe from e

# tests/conftest.py
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

from main import app
from db import Base, lay_session


@pytest_asyncio.fixture
async def session():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with maker() as s:
        yield s

    await engine.dispose()


@pytest_asyncio.fixture
async def client(session):
    async def override():
        yield session

    app.dependency_overrides[lay_session] = override
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

# tests/test_ghi_chu.py
import pytest


@pytest.mark.asyncio
async def test_danh_sach_rong(client):
    r = await client.get("/ghi-chu")
    assert r.status_code == 401        # chưa đăng nhập


@pytest.mark.asyncio
async def test_tao_ghi_chu(client, token_hop_le):
    r = await client.post(
        "/ghi-chu",
        json={"tieu_de": "Việc A", "do_uu_tien": 3},
        headers={"Authorization": f"Bearer {token_hop_le}"},
    )
    assert r.status_code == 201
    assert r.json()["tieu_de"] == "Việc A"


@pytest.mark.asyncio
@pytest.mark.parametrize("payload,ma_loi", [
    ({}, 422),                                        # thiếu tieu_de
    ({"tieu_de": ""}, 422),                           # rỗng
    ({"tieu_de": "X" * 201}, 422),                    # quá dài
    ({"tieu_de": "A", "do_uu_tien": 5}, 422),         # ưu tiên 5 thiếu nội dung
    ({"tieu_de": "A", "do_uu_tien": 99}, 422),        # ngoài khoảng
])
async def test_du_lieu_khong_hop_le(client, token_hop_le, payload, ma_loi):
    r = await client.post("/ghi-chu", json=payload,
                          headers={"Authorization": f"Bearer {token_hop_le}"})
    assert r.status_code == ma_loi

from fastapi import BackgroundTasks


async def gui_email_nhac_viec(email: str, tieu_de: str) -> None:
    async with httpx.AsyncClient() as c:
        await c.post("https://api.email.com/send", json={"to": email, "subject": tieu_de})


@router.post("/{ghi_chu_id}/nhac-viec", status_code=202)
async def nhac_viec(ghi_chu_id: int, tasks: BackgroundTasks,
                    session: SessionDep, user: CurrentUser):
    gc = await lay_mot(session, ghi_chu_id, user.id)
    if gc is None:
        raise HTTPException(404, "Không tìm thấy ghi chú")

    tasks.add_task(gui_email_nhac_viec, user.email, gc.tieu_de)
    return {"thong_diep": "Đã lên lịch gửi nhắc việc"}

from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
import redis.asyncio as redis


@app.on_event("startup")
async def khoi_dong_cache():
    FastAPICache.init(RedisBackend(redis.from_url("redis://localhost")), prefix="api")


@router.get("/thong-ke")
@cache(expire=300)               # cache 5 phút
async def thong_ke(session: SessionDep, user: CurrentUser):
    return await tinh_thong_ke(session, user.id)

import asyncio
import time
import httpx

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


async def tuan_tu() -> list[dict]:
    ket_qua = []
    async with httpx.AsyncClient(timeout=10) as c:
        for coin in COINS:
            r = await c.get(URL, params={"ids": coin, "vs_currencies": "usd"})
            ket_qua.append(r.json())
    return ket_qua


async def dong_thoi() -> list[dict]:
    async with httpx.AsyncClient(timeout=10) as c:
        phan_hoi = await asyncio.gather(
            *(c.get(URL, params={"ids": coin, "vs_currencies": "usd"}) for coin in COINS)
        )
    return [r.json() for r in phan_hoi]


async def main():
    for ham in (tuan_tu, dong_thoi):
        t = time.perf_counter()
        await ham()
        print(f"{ham.__name__:<10}: {time.perf_counter() - t:.2f}s")


asyncio.run(main())

import asyncio
import time
import logging

logger = logging.getLogger(__name__)


class PhatHienBlocking:
    """Cảnh báo khi event loop bị chặn quá lâu."""

    def __init__(self, nguong_ms: float = 200):
        self.nguong = nguong_ms / 1000
        self.tac_vu: asyncio.Task | None = None

    async def bat_dau(self):
        self.tac_vu = asyncio.create_task(self._giam_sat())

    async def ket_thuc(self):
        if self.tac_vu:
            self.tac_vu.cancel()

    async def _giam_sat(self):
        while True:
            truoc = time.perf_counter()
            await asyncio.sleep(0.05)
            do_tre = time.perf_counter() - truoc - 0.05
            if do_tre > self.nguong:
                logger.warning(
                    "⚠️ Event loop bị chặn %.0fms — kiểm tra code blocking!",
                    do_tre * 1000,
                )

from fastapi.concurrency import run_in_threadpool
from passlib.context import CryptContext

pwd = CryptContext(schemes=["bcrypt"])


@app.post("/dang-ky")
async def dang_ky(du_lieu: NguoiDungVao, session: SessionDep):
    # bcrypt tốn ~250ms CPU — phải đẩy ra threadpool, đừng chặn event loop
    mat_khau_bam = await run_in_threadpool(pwd.hash, du_lieu.mat_khau)
    ...

import cProfile
import pstats
import io


def profile_ham(ham, *args, **kwargs):
    pr = cProfile.Profile()
    pr.enable()
    ket_qua = ham(*args, **kwargs)
    pr.disable()

    s = io.StringIO()
    ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
    ps.print_stats(15)
    print(s.getvalue())
    return ket_qua

import logging
import sys
import uuid
from contextvars import ContextVar

from fastapi import Request

request_id: ContextVar[str] = ContextVar("request_id", default="-")


class BoThemRequestId(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id.get()
        return True


def cau_hinh_logging():
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(logging.Formatter(
        '{"time":"%(asctime)s","level":"%(levelname)s",'
        '"logger":"%(name)s","request_id":"%(request_id)s","msg":"%(message)s"}'
    ))
    handler.addFilter(BoThemRequestId())

    root = logging.getLogger()
    root.handlers.clear()
    root.addHandler(handler)
    root.setLevel(logging.INFO)


@app.middleware("http")
async def them_request_id(request: Request, call_next):
    ma = request.headers.get("X-Request-ID") or uuid.uuid4().hex[:12]
    request_id.set(ma)
    response = await call_next(request)
    response.headers["X-Request-ID"] = ma
    return response

from prometheus_fastapi_instrumentator import Instrumentator

Instrumentator().instrument(app).expose(app, endpoint="/metrics")


@app.get("/suc-khoe", tags=["Hệ thống"])
async def suc_khoe(session: SessionDep):
    """Health check cho load balancer — kiểm tra cả database."""
    try:
        await session.execute(text("SELECT 1"))
    except Exception as e:
        raise HTTPException(503, f"Database không phản hồi: {e}") from e
    return {"trang_thai": "ok"}