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

# JSON request/response về bản chất là dict
du_lieu = {"ten": "Minh", "tuoi": 25, "so_thich": ["đọc sách", "chạy bộ"]}

# Các thao tác phải thành thạo
du_lieu.get("email", "chưa có")           # lấy có giá trị mặc định
du_lieu.setdefault("diem", 0)             # gán nếu chưa tồn tại
{**du_lieu, "tuoi": 26}                   # tạo dict mới, ghi đè
{k: v for k, v in du_lieu.items() if k != "so_thich"}   # dict comprehension

from functools import wraps
import time
import logging

logger = logging.getLogger(__name__)


def do_thoi_gian(f):
    """Decorator đo thời gian chạy — bạn sẽ tự viết nhiều cái như thế này."""
    @wraps(f)
    def wrapper(*args, **kwargs):
        bat_dau = time.perf_counter()
        try:
            return f(*args, **kwargs)
        finally:
            thoi_gian = time.perf_counter() - bat_dau
            logger.info("%s chạy mất %.4fs", f.__name__, thoi_gian)
    return wrapper


@do_thoi_gian
def xu_ly_du_lieu(ban_ghi):
    return [r for r in ban_ghi if r["active"]]

from pathlib import Path
import json

CAU_HINH = Path("config.json")


def doc_cau_hinh() -> dict:
    if not CAU_HINH.exists():
        return {}
    try:
        return json.loads(CAU_HINH.read_text(encoding="utf-8"))
    except json.JSONDecodeError as e:
        raise ValueError(f"config.json không hợp lệ: {e}") from e

# app.py
from flask import Flask, render_template, request, redirect, url_for, flash, session

app = Flask(__name__)
app.secret_key = "đọc-từ-env-trong-thực-tế"

GHI_CHU = []


@app.route("/")
def trang_chu():
    return render_template("index.html", ghi_chu=GHI_CHU)


@app.route("/them", methods=["GET", "POST"])
def them_ghi_chu():
    if request.method == "POST":
        noi_dung = request.form.get("noi_dung", "").strip()

        if not noi_dung:
            flash("Bạn chưa nhập nội dung", "loi")
        elif len(noi_dung) > 500:
            flash("Nội dung quá dài (tối đa 500 ký tự)", "loi")
        else:
            GHI_CHU.append({"id": len(GHI_CHU) + 1, "noi_dung": noi_dung})
            flash("Đã thêm ghi chú", "thanh_cong")
            return redirect(url_for("trang_chu"))

    return render_template("them.html")

import os
from pathlib import Path

GOC = Path(__file__).resolve().parent.parent


class Config:
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-key-doi-trong-production")
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JSON_SORT_KEYS = False


class DevConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL", "sqlite:///dev.db")


class TestConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"


class ProdConfig(Config):
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.environ["DATABASE_URL"]      # bắt buộc có
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_HTTPONLY = True
    SESSION_COOKIE_SAMESITE = "Lax"


CAU_HINH = {"dev": DevConfig, "test": TestConfig, "prod": ProdConfig}

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_cors import CORS

db = SQLAlchemy()
migrate = Migrate()
cors = CORS()

import os
import logging
from flask import Flask

from .config import CAU_HINH
from .extensions import db, migrate, cors


def create_app(moi_truong: str | None = None) -> Flask:
    """Application factory — luôn dùng pattern này."""
    app = Flask(__name__)
    moi_truong = moi_truong or os.environ.get("FLASK_ENV", "dev")
    app.config.from_object(CAU_HINH[moi_truong])

    db.init_app(app)
    migrate.init_app(app, db)
    cors.init_app(app, resources={r"/api/*": {"origins": "*"}})

    from .blueprints.main import bp as main_bp
    from .blueprints.api import bp as api_bp
    app.register_blueprint(main_bp)
    app.register_blueprint(api_bp, url_prefix="/api")

    cau_hinh_logging(app)
    dang_ky_error_handler(app)

    return app


def cau_hinh_logging(app: Flask) -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
    )


def dang_ky_error_handler(app: Flask) -> None:
    @app.errorhandler(404)
    def khong_tim_thay(e):
        return {"loi": "Không tìm thấy tài nguyên"}, 404

    @app.errorhandler(500)
    def loi_server(e):
        app.logger.exception("Lỗi 500")
        return {"loi": "Lỗi hệ thống"}, 500

from flask import Blueprint, jsonify

bp = Blueprint("api", __name__)


@bp.get("/suc-khoe")
def suc_khoe():
    return jsonify({"trang_thai": "ok"})


@bp.get("/ghi-chu")
def danh_sach():
    return jsonify({"ghi_chu": []})

from datetime import datetime, timezone
from sqlalchemy import String, Text, Boolean, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column

from ..extensions import db


class GhiChu(db.Model):
    __tablename__ = "ghi_chu"

    id: Mapped[int] = mapped_column(primary_key=True)
    tieu_de: Mapped[str] = mapped_column(String(200), nullable=False, index=True)
    noi_dung: Mapped[str] = mapped_column(Text, default="")
    da_xong: Mapped[bool] = mapped_column(Boolean, default=False, index=True)
    ngay_tao: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    ngay_cap_nhat: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), onupdate=lambda: datetime.now(timezone.utc)
    )

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "tieu_de": self.tieu_de,
            "noi_dung": self.noi_dung,
            "da_xong": self.da_xong,
            "ngay_tao": self.ngay_tao.isoformat() if self.ngay_tao else None,
        }

    def __repr__(self) -> str:
        return f"<GhiChu {self.id} {self.tieu_de!r}>"

from sqlalchemy import select
from .extensions import db
from .models.ghi_chu import GhiChu


def lay_tat_ca(chi_chua_xong: bool = False) -> list[GhiChu]:
    stmt = select(GhiChu).order_by(GhiChu.ngay_tao.desc())
    if chi_chua_xong:
        stmt = stmt.where(GhiChu.da_xong.is_(False))
    return list(db.session.scalars(stmt))


def tim_theo_tu_khoa(tu_khoa: str) -> list[GhiChu]:
    stmt = select(GhiChu).where(GhiChu.tieu_de.ilike(f"%{tu_khoa}%"))
    return list(db.session.scalars(stmt))

# app/blueprints/api.py
from flask import Blueprint, jsonify, request
from marshmallow import Schema, fields, validate, ValidationError

from ..extensions import db
from ..models.ghi_chu import GhiChu

bp = Blueprint("api", __name__)


class GhiChuSchema(Schema):
    tieu_de = fields.Str(required=True, validate=validate.Length(min=1, max=200))
    noi_dung = fields.Str(load_default="", validate=validate.Length(max=5000))
    da_xong = fields.Bool(load_default=False)


schema = GhiChuSchema()
schema_nhieu = GhiChuSchema(many=True)


@bp.get("/ghi-chu")
def danh_sach():
    chi_chua_xong = request.args.get("chua_xong", "").lower() in ("1", "true", "yes")
    try:
        gioi_han = min(int(request.args.get("gioi_han", 50)), 200)
    except ValueError:
        return jsonify({"loi": "gioi_han phải là số nguyên"}), 400

    ds = lay_tat_ca(chi_chua_xong)[:gioi_han]
    return jsonify({"du_lieu": schema_nhieu.dump(ds), "so_luong": len(ds)})


@bp.get("/ghi-chu/<int:ghi_chu_id>")
def chi_tiet(ghi_chu_id: int):
    gc = db.session.get(GhiChu, ghi_chu_id)
    if gc is None:
        return jsonify({"loi": f"Không tìm thấy ghi chú id={ghi_chu_id}"}), 404
    return jsonify(schema.dump(gc))


@bp.post("/ghi-chu")
def tao_moi():
    try:
        du_lieu = schema.load(request.get_json(silent=True) or {})
    except ValidationError as e:
        return jsonify({"loi": "Dữ liệu không hợp lệ", "chi_tiet": e.messages}), 422

    gc = GhiChu(**du_lieu)
    db.session.add(gc)
    db.session.commit()
    return jsonify(schema.dump(gc)), 201


@bp.put("/ghi-chu/<int:ghi_chu_id>")
def cap_nhat(ghi_chu_id: int):
    gc = db.session.get(GhiChu, ghi_chu_id)
    if gc is None:
        return jsonify({"loi": "Không tìm thấy"}), 404

    try:
        du_lieu = schema.load(request.get_json(silent=True) or {}, partial=True)
    except ValidationError as e:
        return jsonify({"loi": "Dữ liệu không hợp lệ", "chi_tiet": e.messages}), 422

    for khoa, gia_tri in du_lieu.items():
        setattr(gc, khoa, gia_tri)
    db.session.commit()
    return jsonify(schema.dump(gc))


@bp.delete("/ghi-chu/<int:ghi_chu_id>")
def xoa(ghi_chu_id: int):
    gc = db.session.get(GhiChu, ghi_chu_id)
    if gc is None:
        return jsonify({"loi": "Không tìm thấy"}), 404
    db.session.delete(gc)
    db.session.commit()
    return "", 204

from flask_login import UserMixin, login_required, current_user
from werkzeug.security import generate_password_hash, check_password_hash


class NguoiDung(db.Model, UserMixin):
    __tablename__ = "nguoi_dung"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(120), unique=True, index=True)
    mat_khau_hash: Mapped[str] = mapped_column(String(255))
    vai_tro: Mapped[str] = mapped_column(String(20), default="user")

    def dat_mat_khau(self, mat_khau: str) -> None:
        self.mat_khau_hash = generate_password_hash(mat_khau, method="scrypt")

    def kiem_tra_mat_khau(self, mat_khau: str) -> bool:
        return check_password_hash(self.mat_khau_hash, mat_khau)

    @property
    def la_admin(self) -> bool:
        return self.vai_tro == "admin"

# ⚠️ LỖI KINH ĐIỂN — không bao giờ làm thế này
@app.post("/api/xoa")
def xoa():
    du_lieu = request.get_json()
    if du_lieu.get("vai_tro") == "admin":      # ❌ client tự khai mình là admin!
        ...

import pytest
from app import create_app
from app.extensions import db as _db


@pytest.fixture(scope="session")
def app():
    app = create_app("test")
    with app.app_context():
        _db.create_all()
        yield app
        _db.drop_all()


@pytest.fixture
def client(app):
    return app.test_client()


@pytest.fixture(autouse=True)
def sach_database(app):
    """Mỗi test chạy trên database sạch."""
    yield
    with app.app_context():
        _db.session.rollback()
        _db.drop_all()
        _db.create_all()

def test_danh_sach_rong(client):
    r = client.get("/api/ghi-chu")
    assert r.status_code == 200
    assert r.get_json()["so_luong"] == 0


def test_tao_ghi_chu(client):
    r = client.post("/api/ghi-chu", json={"tieu_de": "Việc cần làm"})
    assert r.status_code == 201
    assert r.get_json()["tieu_de"] == "Việc cần làm"


def test_tao_thieu_tieu_de(client):
    r = client.post("/api/ghi-chu", json={"noi_dung": "abc"})
    assert r.status_code == 422
    assert "tieu_de" in r.get_json()["chi_tiet"]


def test_lay_khong_ton_tai(client):
    r = client.get("/api/ghi-chu/99999")
    assert r.status_code == 404


def test_xoa_thanh_cong(client):
    tao = client.post("/api/ghi-chu", json={"tieu_de": "X"}).get_json()
    r = client.delete(f"/api/ghi-chu/{tao['id']}")
    assert r.status_code == 204
    assert client.get(f"/api/ghi-chu/{tao['id']}").status_code == 404

from app import create_app

app = create_app("prod")

if __name__ == "__main__":
    app.run()

import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path


def cau_hinh_logging(app):
    """Cấu hình logging: vừa ra console vừa ghi file xoay vòng."""
    dinh_dang = logging.Formatter(
        "%(asctime)s | %(levelname)-8s | %(name)s:%(lineno)d | %(message)s"
    )

    console = logging.StreamHandler()
    console.setFormatter(dinh_dang)

    Path("logs").mkdir(exist_ok=True)
    file_handler = RotatingFileHandler(
        "logs/app.log", maxBytes=5_000_000, backupCount=5, encoding="utf-8"
    )
    file_handler.setFormatter(dinh_dang)

    app.logger.handlers.clear()
    app.logger.addHandler(console)
    app.logger.addHandler(file_handler)
    app.logger.setLevel(logging.DEBUG if app.config["DEBUG"] else logging.INFO)

    # Giảm ồn từ thư viện bên thứ ba
    logging.getLogger("werkzeug").setLevel(logging.WARNING)
    logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)

import time
import uuid
from flask import g, request


@app.before_request
def bat_dau_do():
    g.ma_request = uuid.uuid4().hex[:8]
    g.thoi_diem_bat_dau = time.perf_counter()


@app.after_request
def ket_thuc_do(response):
    thoi_gian = (time.perf_counter() - g.thoi_diem_bat_dau) * 1000
    response.headers["X-Request-ID"] = g.ma_request
    app.logger.info(
        "[%s] %s %s -> %s (%.1f ms)",
        g.ma_request, request.method, request.path, response.status_code, thoi_gian,
    )
    if thoi_gian > 500:
        app.logger.warning("[%s] Request chậm: %s", g.ma_request, request.path)
    return response

# Giới hạn kích thước upload — bảo vệ server khỏi bị làm nghẽn
app.config["MAX_CONTENT_LENGTH"] = 16 * 1024 * 1024      # 16 MB

# ❌ Route làm quá nhiều việc
@bp.post("/don-hang")
def tao_don_hang():
    du_lieu = request.get_json()

    # validate
    if not du_lieu.get("khach_hang_id"):
        return jsonify({"loi": "Thiếu khách hàng"}), 400

    # kiểm tra tồn kho
    for item in du_lieu["items"]:
        sp = db.session.get(SanPham, item["san_pham_id"])
        if sp is None or sp.ton_kho < item["so_luong"]:
            return jsonify({"loi": f"Hết hàng: {item['san_pham_id']}"}), 409

    # tính tiền
    tong = sum(item["so_luong"] * item["don_gia"] for item in du_lieu["items"])
    giam_gia = tinh_giam_gia(tong, du_lieu.get("ma_giam_gia"))

    # tạo bản ghi
    dh = DonHang(tong_tien=tong - giam_gia, khach_hang_id=du_lieu["khach_hang_id"])
    db.session.add(dh)
    db.session.flush()

    # trừ tồn kho
    for item in du_lieu["items"]:
        sp = db.session.get(SanPham, item["san_pham_id"])
        sp.ton_kho -= item["so_luong"]

    db.session.commit()

    # gửi email
    gui_email_xac_nhan(dh)

    return jsonify(dh.to_dict()), 201

from dataclasses import dataclass
from decimal import Decimal
from sqlalchemy import select, update
from sqlalchemy.orm import Session

from ..extensions import db
from ..models.don_hang import DonHang, ChiTietDonHang
from ..models.san_pham import SanPham
from .exceptions import HetHangError, KhachHangKhongTonTai


@dataclass
class MucDonHang:
    san_pham_id: int
    so_luong: int
    don_gia: Decimal


class DichVuDonHang:
    """Toàn bộ nghiệp vụ đơn hàng nằm ở đây — không phụ thuộc Flask."""

    def __init__(self, session: Session):
        self.session = session

    def tao_don_hang(self, khach_hang_id: int, items: list[MucDonHang],
                     ma_giam_gia: str | None = None) -> DonHang:
        if not items:
            raise ValueError("Đơn hàng phải có ít nhất một sản phẩm")

        with self.session.begin_nested():        # savepoint — rollback riêng nếu lỗi
            self._kiem_tra_ton_kho(items)

            tong = sum(m.so_luong * m.don_gia for m in items)
            giam_gia = self._tinh_giam_gia(tong, ma_giam_gia)

            dh = DonHang(khach_hang_id=khach_hang_id, tong_tien=tong - giam_gia)
            self.session.add(dh)
            self.session.flush()

            for muc in items:
                self.session.add(ChiTietDonHang(
                    don_hang_id=dh.id, san_pham_id=muc.san_pham_id,
                    so_luong=muc.so_luong, don_gia=muc.don_gia,
                ))
                self.session.execute(
                    update(SanPham)
                    .where(SanPham.id == muc.san_pham_id)
                    .values(ton_kho=SanPham.ton_kho - muc.so_luong)
                )

        return dh

    def _kiem_tra_ton_kho(self, items: list[MucDonHang]) -> None:
        for muc in items:
            sp = self.session.get(SanPham, muc.san_pham_id)
            if sp is None or sp.ton_kho < muc.so_luong:
                raise HetHangError(muc.san_pham_id, muc.so_luong)

    def _tinh_giam_gia(self, tong: Decimal, ma: str | None) -> Decimal:
        if not ma:
            return Decimal(0)
        # gọi repository mã giảm giá...
        return Decimal(0)

@bp.post("/don-hang")
@yeu_cau_dang_nhap
def tao_don_hang():
    """HTTP: parse → gọi service → format response. Không chứa nghiệp vụ."""
    try:
        du_lieu = schema.load(request.get_json(silent=True) or {})
    except ValidationError as e:
        return jsonify({"loi": "Dữ liệu không hợp lệ", "chi_tiet": e.messages}), 422

    dich_vu = DichVuDonHang(db.session)
    try:
        dh = dich_vu.tao_don_hang(
            khach_hang_id=g.nguoi_dung.id,
            items=[MucDonHang(**m) for m in du_lieu["items"]],
            ma_giam_gia=du_lieu.get("ma_giam_gia"),
        )
        db.session.commit()
    except HetHangError as e:
        db.session.rollback()
        return jsonify({"loi": str(e)}), 409

    gui_email_xac_nhan(dh)          # hoặc đẩy vào Celery
    return jsonify(dh.to_dict()), 201

# Test thuần Python — không cần Flask, không cần HTTP
def test_tao_don_hang_het_hang(session):
    sp = SanPham(ten="Bàn phím", gia=Decimal("250000"), ton_kho=1)
    session.add(sp)
    session.flush()

    dich_vu = DichVuDonHang(session)
    with pytest.raises(HetHangError):
        dich_vu.tao_don_hang(1, [MucDonHang(sp.id, 5, Decimal("250000"))])