7 bài viết được gắn thẻ "Lộ trình Python"

import random

so_bi_mat = random.randint(1, 100)
so_lan_doan = 0

print("Mình đang nghĩ một số từ 1 đến 100. Bạn đoán thử xem!")

while True:
    doan = int(input("Số bạn đoán: "))
    so_lan_doan += 1

    if doan < so_bi_mat:
        print("⬆️ Cao hơn nữa!")
    elif doan > so_bi_mat:
        print("⬇️ Thấp hơn nữa!")
    else:
        print(f"🎉 Đúng rồi! Bạn đoán {so_lan_doan} lần.")
        break

import turtle

man = turtle.Turtle()
man.speed(0)

for i in range(100):
    man.forward(i * 2)
    man.right(59)

turtle.done()

class HocSinh:
    def __init__(self, ten, lop):
        self.ten = ten
        self.lop = lop
        self.diem = {}

    def them_diem(self, mon, diem):
        if not 0 <= diem <= 10:
            raise ValueError("Điểm phải từ 0 đến 10")
        self.diem[mon] = diem
        return self

    @property
    def diem_trung_binh(self):
        if not self.diem:
            return 0
        return sum(self.diem.values()) / len(self.diem)

    @property
    def xep_loai(self):
        tb = self.diem_trung_binh
        if tb >= 8.0:
            return "Giỏi"
        if tb >= 6.5:
            return "Khá"
        if tb >= 5.0:
            return "Trung bình"
        return "Yếu"

    def __str__(self):
        return f"{self.ten} ({self.lop}) — TB {self.diem_trung_binh:.2f} — {self.xep_loai}"


hs = HocSinh("Minh", "12A1")
hs.them_diem("Toán", 9).them_diem("Lý", 8).them_diem("Hoá", 7.5)
print(hs)

from abc import ABC, abstractmethod


class Hinh(ABC):
    @abstractmethod
    def dien_tich(self):
        ...

    def __str__(self):
        return f"{type(self).__name__} — diện tích {self.dien_tich():.2f}"


class HinhTron(Hinh):
    def __init__(self, r):
        self.r = r

    def dien_tich(self):
        return 3.14159 * self.r ** 2


class HinhChuNhat(Hinh):
    def __init__(self, dai, rong):
        self.dai = dai
        self.rong = rong

    def dien_tich(self):
        return self.dai * self.rong


for h in [HinhTron(3), HinhChuNhat(4, 5)]:
    print(h)          # đa hình qua __str__ + dien_tich

import sqlite3

with sqlite3.connect("shop.db") as conn:
    conn.execute("""
        CREATE TABLE IF NOT EXISTS san_pham (
            id INTEGER PRIMARY KEY,
            ten TEXT NOT NULL,
            gia REAL NOT NULL
        )
    """)
    conn.executemany(
        "INSERT INTO san_pham (ten, gia) VALUES (?, ?)",
        [("Bàn phím", 250000), ("Chuột", 150000), ("Màn hình", 2500000)],
    )
    conn.commit()

    # Luôn dùng tham số hoá (?) — KHÔNG nối chuỗi SQL
    for ten, gia in conn.execute("SELECT ten, gia FROM san_pham WHERE gia > ?", (200000,)):
        print(f"{ten}: {gia:,.0f} đ")

import pytest
from shop import tinh_giam_gia


def test_giam_gia_binh_thuong():
    assert tinh_giam_gia(100000, 10) == 90000


def test_giam_gia_bien():
    assert tinh_giam_gia(0, 10) == 0


def test_giam_gia_am_thi_loi():
    with pytest.raises(ValueError):
        tinh_giam_gia(-1000, 10)

# ❌ Sinh viên năm 2
def get_user(id):
    import sqlite3
    conn = sqlite3.connect("db.sqlite")
    cur = conn.execute(f"SELECT * FROM users WHERE id = {id}")
    return cur.fetchone()

# ✅ Sinh viên năm 4
import logging
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class User:
    id: int
    ten: str
    email: str


def lay_user_theo_id(user_id: int) -> User | None:
    """Lấy user theo id, trả None nếu không tồn tại."""
    if user_id <= 0:
        raise ValueError(f"user_id phải dương, nhận {user_id}")

    with get_connection() as conn:
        row = conn.execute(
            "SELECT id, ten, email FROM users WHERE id = ?", (user_id,)
        ).fetchone()

    if row is None:
        logger.info("Không tìm thấy user id=%s", user_id)
        return None

    return User(*row)

from pathlib import Path
import pandas as pd

THU_MUC = Path("bao-cao-2026")
tong_hop = []

for file in sorted(THU_MUC.glob("*.xlsx")):
    df = pd.read_excel(file, sheet_name="Sheet1")
    df["nguon"] = file.stem
    tong_hop.append(df)
    print(f"✓ {file.name}: {len(df)} dòng")

if not tong_hop:
    raise SystemExit("Không tìm thấy file nào!")

ket_qua = pd.concat(tong_hop, ignore_index=True)
ket_qua = ket_qua.drop_duplicates(subset=["ma_don", "ngay"])

tong_theo_thang = (
    ket_qua.assign(thang=pd.to_datetime(ket_qua["ngay"]).dt.to_period("M"))
           .groupby("thang")
           .agg(so_don=("ma_don", "nunique"), doanh_thu=("so_tien", "sum"))
           .reset_index()
)

tong_theo_thang.to_excel("tong-hop.xlsx", index=False)
print(f"\n✅ Đã tổng hợp {len(ket_qua)} dòng → tong-hop.xlsx")

import time
import numpy as np
from functools import lru_cache

# ❌ Tính lặp lại — chậm
def tinh_tong_cham(n):
    return sum(i * i for i in range(n))

# ✅ Cache khi hàm thuần tuý và gọi nhiều lần
@lru_cache(maxsize=128)
def fib(n):
    return n if n < 2 else fib(n - 1) + fib(n - 2)

# ✅ Vector hoá — nhanh hơn vòng lặp Python hàng chục lần
def tinh_tong_nhanh(n):
    arr = np.arange(n)
    return int((arr * arr).sum())

for ham, ten in [(tinh_tong_cham, "vòng lặp"), (tinh_tong_nhanh, "numpy")]:
    t = time.perf_counter()
    ham(1_000_000)
    print(f"{ten}: {time.perf_counter() - t:.4f}s")

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