# 若未安装 Homebrew，先到官网 https://brew.sh 复制最新安装命令执行
# 安装 Python 3.12
brew install python@3.12

sudo apt update
sudo apt install python3.12 python3.12-venv python3-pip -y

python --version

# 进入你的项目目录（示例）
mkdir my_project && cd my_project
# 创建虚拟环境（会在当前目录生成 .venv 文件夹）
python -m venv .venv
# 激活虚拟环境 —— Windows PowerShell
.venv\Scripts\Activate.ps1
# 激活虚拟环境 —— macOS / Linux
source .venv/bin/activate

(.venv) PS D:\my_project>

deactivate

# 一次性使用
pip install fastapi -i https://pypi.tuna.tsinghua.edu.cn/simple
# 永久配置（写入用户配置文件）
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

fastapi==0.115.0
uvicorn[standard]==0.30.6
sqlalchemy==2.0.35
pymysql==1.1.1

{
    "editor.formatOnSave": true,
    "editor.codeActionsOnSave": {
        "source.fixAll.ruff": "explicit",
        "source.organizeImports.ruff": "explicit"
    },
    "[python]": {
        "editor.defaultFormatter": "charliermarsh.ruff"
    }
}

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python: 当前文件",
            "type": "debugpy",
            "request": "launch",
            "program": "${file}",
            "console": "integratedTerminal"
        }
    ]
}

pip install ruff

[project]
name = "my_project"
version = "0.1.0"
requires-python = ">=3.12"
[tool.ruff]
line-length = 100
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "W", "I"]  # 基础错误、pyflakes、警告、import 排序
[tool.ruff.format]
quote-style = "double"

# 检查问题
ruff check .
# 自动修复能修的（import 排序、未使用 import 等）
ruff check . --fix
# 格式化整个项目
ruff format .

"""data_structures.py — 常用数据结构操作速查"""
# ---------- 列表 ----------
nums = [3, 1, 4, 1, 5, 9, 2, 6]
nums.append(5)          # 尾部追加
nums.insert(0, 100)     # 指定位置插入
nums.remove(100)        # 按值删除（只删第一个匹配）
top = nums.pop()        # 弹出末尾元素
nums.sort(reverse=True)  # 原地排序
print(nums, top)
# 切片：[起:止:步长]，含头不含尾
print(nums[1:4])      # 第 2 到第 4 个
print(nums[::-1])     # 反转列表的经典写法
print(nums[:3])       # 前 3 个
# ---------- 字典 ----------
user = {"name": "小明", "age": 18}
user["city"] = "北京"                  # 新增
age = user.get("score", 0)              # 安全取值，键不存在返回默认 0
val = user.pop("age", None)             # 删除并返回，键不存在返回 None
print(user, age, val)
# 遍历的最佳方式：同时拿键和值
for key, value in user.items():
    print(f"{key} = {value}")
# 合并字典（3.9+ 语法）
defaults = {"theme": "light", "page_size": 20}
custom = {"theme": "dark"}
merged = defaults | custom
print(merged)  # {'theme': 'dark', 'page_size': 20}
# ---------- 集合 ----------
a = {1, 2, 3, 4}
b = {3, 4, 5, 6}
print(a & b)   # 交集 {3, 4}
print(a | b)   # 并集 {1, 2, 3, 4, 5, 6}
print(a - b)   # 差集 {1, 2}
# 去重三连：列表 -> 集合 -> 列表（不保序则用 set，保序见下方）
data = [1, 3, 2, 3, 1]
print(list(set(data)))

"""comprehension.py — 推导式"""
nums = [1, 2, 3, 4, 5, 6, 7, 8]
# 列表推导式：平方
squares = [n * n for n in nums]
print(squares)
# 带条件过滤：只要偶数
evens = [n for n in nums if n % 2 == 0]
print(evens)
# 带条件表达式：偶数原样，奇数变 0
mapped = [n if n % 2 == 0 else 0 for n in nums]
print(mapped)
# 字典推导式：单词 -> 长度
words = ["apple", "banana", "cherry"]
lengths = {w: len(w) for w in words}
print(lengths)
# 集合推导式：所有单词的首字母
initials = {w[0] for w in words}
print(initials)
# 生成器表达式：外面是圆括号，惰性求值不占内存
total = sum(n * n for n in range(1_000_000))
print(total)

"""unpack.py — 解包与格式化"""
# 多变量同时赋值
x, y = 10, 20
x, y = y, x                     # Python 交换两数不需要中间变量
print(x, y)
# 星号解包：收集中间所有元素
first, *middle, last = [1, 2, 3, 4, 5]
print(first, middle, last)      # 1 [2, 3, 4] 5
# 函数返回元组直接解包
def min_max(values):
    return min(values), max(values)
lo, hi = min_max([3, 1, 4, 1, 5])
print(lo, hi)
# f-string：字符串格式化的唯一推荐方式
name, score = "小明", 92.5
print(f"{name} 的成绩是 {score}")        # 直接嵌入变量
print(f"及格率 {0.856:.1%}")             # 百分比保留 1 位
print(f"{1234567:,}")                    # 千分位 1,234,567
print(f"{name!r}")                       # 带引号输出，调试利器

"""func_args.py — 函数参数全解"""
def create_user(name, age=18, *tags, city, is_vip=False, **extra):
    """参数顺序固定为：位置参数 -> *args -> 关键字参数 -> **kwargs"""
    print(f"name={name}, age={age}, tags={tags}, city={city}, "
          f"is_vip={is_vip}, extra={extra}")
create_user("小明", 20, "admin", "beta", city="北京", remark="测试")
# *args 把多余的“位置参数”打包成元组：(admin, beta)
# **kwargs 把多余的“关键字参数”打包成字典：{remark: 测试}
# city 没有默认值，调用时必须传，否则报 TypeError

def add_item(item, items=None):
    if items is None:
        items = []
    items.append(item)
    return items

"""first_class.py — 函数是一等公民"""
def apply(func, value):
    """把函数作为参数执行"""
    return func(value)
print(apply(abs, -10))          # 10
print(apply(str.upper, "abc"))  # ABC
# lambda：一行的小函数，常作为排序 key
users = [
    {"name": "小明", "age": 22},
    {"name": "小红", "age": 19},
    {"name": "老王", "age": 35},
]
users_by_age = sorted(users, key=lambda u: u["age"])
print([u["name"] for u in users_by_age])   # 小红 小明 老王
# key 还可以返回元组，实现多级排序
rows = [("b", 2), ("a", 3), ("b", 1)]
print(sorted(rows, key=lambda r: (r[0], -r[1])))

"""typing_basic.py — 类型注解"""
def total_price(price: float, count: int = 1) -> float:
    """参数类型: 参数名后加冒号；返回类型: 括号后加 ->"""
    return price * count
# 变量注解（通常不必写，IDE 能推断；写清楚有自文档价值）
timeout: int = 30
db_url: str = "mysql://localhost/mydb"
# 容器类型
names: list[str] = []
scores: dict[str, int] = {}
# 可选类型：值可能是 None
def find_user(user_id: int) -> str | None:
    if user_id == 1:
        return "小明"
    return None

pip install mypy
mypy typing_basic.py

"""decorator.py — 装饰器入门到实战"""
import time
import functools
def timer(func):
    """计时装饰器"""
    @functools.wraps(func)   # 保住原函数的名字和文档
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        cost = (time.perf_counter() - start) * 1000
        print(f"[{func.__name__}] 耗时 {cost:.2f} ms")
        return result
    return wrapper
@timer
def download_data(size: int) -> list[int]:
    return list(range(size))
download_data(1_000_000)
# 输出: [download_data] 耗时 23.31 ms
@timer 等价于 download_data = timer(download_data)：原函数被替换成包装后的 wrapper，功能增强而调用方式不变。

"""retry_decorator.py — 带参数的重试装饰器"""
import functools
import time
def retry(times: int = 3, delay: float = 1.0):
    """失败自动重试 times 次"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == times:
                        raise   # 最后一次失败，把异常原样抛出
                    print(f"第 {attempt} 次失败: {e}，{delay}s 后重试")
                    time.sleep(delay)
        return wrapper
    return decorator
@retry(times=3, delay=0.5)
def call_flaky_api():
    import random
    if random.random() < 0.7:
        raise ConnectionError("网络抖动")
    return {"ok": True}
print(call_flaky_api())

"""generator.py — 生成器"""
def read_large_file(path: str, chunk_lines: int = 1000):
    """逐块读大文件，内存中永远只有一小块"""
    with open(path, encoding="utf-8") as f:
        batch = []
        for line in f:
            batch.append(line.rstrip())
            if len(batch) >= chunk_lines:
                yield batch   # 暂停并交出一批数据
                batch = []
        if batch:
            yield batch
# 生成器表达式 + sum：一行统计 10 亿数字的和也不会撑爆内存
total = sum(i * i for i in range(1_000_000_000))
print(total)
# 无限序列：生成器不预生成数据，所以“无限”是安全的
def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b
from itertools import islice
print(list(islice(fibonacci(), 10)))  # 只取前 10 个

"""context_manager.py — 自定义上下文管理器"""
class Timer:
    """进入时计时，退出时打印耗时——with 里的代码块计时器"""
    def __enter__(self):
        import time
        self.start = time.perf_counter()
        return self                     # as 后面拿到的就是这个对象
    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        self.cost = (time.perf_counter() - self.start) * 1000
        print(f"代码块耗时 {self.cost:.2f} ms")
        return False                    # False 表示不吞异常，继续向外抛
with Timer():
    total = sum(range(1_000_000))
print("继续执行")

from contextlib import contextmanager
@contextmanager
def open_db_conn(url: str):
    print(f"连接数据库 {url}")
    try:
        yield "connection_object"       # as 拿到的值
    finally:
        print("关闭数据库连接")          # 无论成败都执行
with open_db_conn("mysql://localhost") as conn:
    print(f"使用 {conn} 查询数据")

"""exceptions.py — 异常处理规范"""
class BusinessError(Exception):
    """自定义业务异常：企业项目都会定义自己的异常基类"""
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message
        super().__init__(message)
def withdraw(balance: float, amount: float) -> float:
    if amount <= 0:
        raise BusinessError(400, "取款金额必须大于 0")
    if amount > balance:
        raise BusinessError(4002, "余额不足")
    return balance - amount
try:
    new_balance = withdraw(100, 500)
except BusinessError as e:
    print(f"业务失败: code={e.code}, msg={e.message}")
except Exception as e:                     # 兜底：未知异常
    print(f"系统异常: {e}")
    raise                                  # 兜底后重新抛出，让上层记录日志
else:
    print(f"成功，新余额 {new_balance}")
finally:
    print("无论成败都执行（常用于清理资源）")

"""stdlib_tour.py — 高频标准库一网打尽"""
# ---------- pathlib：现代路径操作（代替 os.path）----------
from pathlib import Path
config_dir = Path("config")
config_dir.mkdir(exist_ok=True)             # 存在也不报错
config_file = config_dir / "app.json"       # 用 / 拼接路径
config_file.write_text('{"debug": true}', encoding="utf-8")
print(config_file.read_text(encoding="utf-8"))
print(config_file.suffix, config_file.exists())  # .json True
# ---------- datetime：时间处理 ----------
from datetime import datetime, timedelta, timezone
now = datetime.now()
utc_now = datetime.now(timezone.utc)        # 服务器代码一律用 UTC
tomorrow = now + timedelta(days=1)
print(now.strftime("%Y-%m-%d %H:%M:%S"))    # 格式化输出
created = datetime.strptime("2026-09-25", "%Y-%m-%d")   # 字符串转时间
print((now - created).days)
# ---------- json：与 API 打交道的基石 ----------
import json
payload = {"name": "小明", "roles": ["admin"], "active": True}
text = json.dumps(payload, ensure_ascii=False)   # 转 JSON 字符串
print(text)
data = json.loads(text)                          # 解析回对象
print(data["roles"])
# ---------- collections：好用的数据容器 ----------
from collections import Counter, defaultdict
words = ["apple", "banana", "apple", "cherry", "apple"]
print(Counter(words).most_common(2))         # [('apple', 3), ('banana', 1)]
groups = defaultdict(list)
for w in words:
    groups[w[0]].append(w)                   # 不存在也不会 KeyError
print(dict(groups))
# ---------- itertools：迭代工具箱 ----------
from itertools import chain, product
print(list(chain([1, 2], [3, 4])))           # 拼接可迭代对象
print(list(product("AB", "12")))             # 笛卡尔积

"""oop_basic.py — 类基础回顾"""
class User:
    role = "member"                     # 类属性：所有实例共享
    def __init__(self, name: str, age: int):
        self.name = name                # 实例属性：每个对象各自一份
        self.age = age
    def greet(self) -> str:            # 实例方法：第一个参数固定是 self
        return f"我是 {self.name}，{self.age} 岁"
class Admin(User):                      # 继承：Admin 自动拥有 User 的一切
    role = "admin"
    def delete_user(self, name: str) -> str:
        return f"管理员 {self.name} 删除了用户 {name}"
u = User("小明", 22)
a = Admin("老王", 35)
print(u.greet(), "|", a.greet())
print(a.delete_user("张三"))
print(isinstance(a, User))              # True：Admin 也是 User

"""dataclass_demo.py — dataclass 全解"""
from dataclasses import dataclass, field, asdict
@dataclass
class ServerConfig:
    host: str                                   # 无默认值的字段在前
    port: int = 8000                            # 有默认值字段在后
    debug: bool = False
    tags: list[str] = field(default_factory=list)  # 可变默认值的标准写法
    def is_local(self) -> bool:                 # 照常定义普通方法
        return self.host in ("localhost", "127.0.0.1")
cfg = ServerConfig(host="127.0.0.1", port=9000, tags=["api", "prod"])
print(cfg)                     # 自动生成的漂亮 repr
print(cfg == ServerConfig("127.0.0.1", 9000, False, ["api", "prod"]))  # True
print(asdict(cfg))             # 转字典：方便写日志、序列化
print(cfg.is_local())

"""property_demo.py — property 与只读属性"""
class Account:
    def __init__(self):
        self._balance = 0.0          # 下划线开头 = 内部变量（约定俗成）
    @property
    def balance(self) -> float:
        """读取余额"""
        return self._balance
    @balance.setter
    def balance(self, value: float):
        """修改余额前校验"""
        if value < 0:
            raise ValueError("余额不能为负")
        self._balance = value
acc = Account()
acc.balance = 100.0                  # 看似赋值，实际走了 setter 校验
print(acc.balance)                   # 100.0
try:
    acc.balance = -1                 # 触发校验
except ValueError as e:
    print(e)

"""magic_methods.py — 高频魔术方法"""
class Playlist:
    def __init__(self, name: str, songs: list[str]):
        self.name = name
        self.songs = songs
    def __repr__(self):
        """print() / 调试时显示的内容"""
        return f"Playlist({self.name!r}, {len(self)} 首)"
    def __len__(self):
        """len(obj) 的行为"""
        return len(self.songs)
    def __getitem__(self, index: int):
        """支持 obj[i] 下标访问和切片"""
        return self.songs[index]
    def __contains__(self, song: str):
        """支持 in 运算符"""
        return song in self.songs
    def __eq__(self, other):
        """支持 == 比较"""
        return isinstance(other, Playlist) and self.songs == other.songs
pl = Playlist("晨跑歌单", ["七里香", "稻香", "晴天"])
print(pl)                    # Playlist('晨跑歌单', 3 首)
print(len(pl), pl[0])        # 3 七里香
print("晴天" in pl)          # True
print(pl == Playlist("随便", ["七里香", "稻香", "晴天"]))  # True（不看名字）

"""composition.py — 组合优于继承"""
class EmailSender:
    def send(self, to: str, subject: str, body: str) -> None:
        print(f"发送邮件到 {to}：[{subject}] {body[:20]}...")
class OrderService:
    """订单服务需要一个发邮件能力 —— 组合进来，而不是继承"""
    def __init__(self, mailer: EmailSender):
        self.mailer = mailer               # 持有对象，而不是继承行为
    def create_order(self, user_email: str) -> None:
        print("创建订单成功")
        self.mailer.send(user_email, "订单通知", "您的订单已创建")
service = OrderService(EmailSender())
service.create_order("user@example.com")

"""mixin_demo.py — Mixin 模式"""
import json
class ToJSONMixin:
    """任何类混入它，就获得转 JSON 的能力"""
    def to_json(self) -> str:
        return json.dumps(vars(self), ensure_ascii=False, default=str)
class LogMixin:
    """混入后，属性变化自动记日志"""
    def __setattr__(self, key, value):
        print(f"[LOG] {key} -> {value}")
        super().__setattr__(key, value)
class Product(ToJSONMixin, LogMixin):     # 多个能力自由组合
    def __init__(self, name: str, price: float):
        self.name = name
        self.price = price
p = Product("机械键盘", 399.0)            # 每次赋值都会打日志
print(p.to_json())

"""paginator.py — 分页器实战"""
from dataclasses import dataclass
from math import ceil
@dataclass
class Page:
    items: list          # 当前页数据
    total: int           # 总条数
    page: int            # 当前页码
    size: int            # 每页条数
    @property
    def pages(self) -> int:
        """总页数：向上取整"""
        return ceil(self.total / self.size) if self.size else 0
    @property
    def has_next(self) -> bool:
        return self.page < self.pages
    @property
    def next_page(self) -> int | None:
        return self.page + 1 if self.has_next else None
    @property
    def prev_page(self) -> int | None:
        return self.page - 1 if self.page > 1 else None
def paginate(data: list, page: int = 1, size: int = 10) -> Page:
    """对列表分页：真实项目中 data 通常替换为数据库查询结果"""
    if page < 1 or size < 1:
        raise ValueError("page 和 size 必须为正整数")
    start = (page - 1) * size
    return Page(
        items=data[start:start + size],
        total=len(data),
        page=page,
        size=size,
    )
if __name__ == "__main__":
    demo = [f"记录-{i}" for i in range(1, 96)]     # 95 条数据
    result = paginate(demo, page=5, size=10)
    print(f"第 {result.page}/{result.pages} 页，共 {result.total} 条")
    print(f"本页内容: {result.items}")
    print(f"下一页: {result.next_page}，上一页: {result.prev_page}")

docker run -d \
  --name mysql-demo \
  -e MYSQL_ROOT_PASSWORD=root123456 \
  -e MYSQL_DATABASE=taskdb \
  -p 3306:3306 \
  mysql:8.4

# 进入容器内的 mysql 命令行
docker exec -it mysql-demo mysql -uroot -proot123456

-- 每条语句以分号结尾；SQL 关键字习惯大写（不强制）
CREATE DATABASE IF NOT EXISTS taskdb DEFAULT CHARACTER SET utf8mb4;
USE taskdb;
-- 用户表
CREATE TABLE users (
    id          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,  -- 主键：自增 ID
    username    VARCHAR(50)  NOT NULL UNIQUE,               -- 登录名，唯一
    email       VARCHAR(100) NOT NULL UNIQUE,
    password_hash VARCHAR(255) NOT NULL,                    -- 存哈希，绝不存明文
    is_active   TINYINT(1)   NOT NULL DEFAULT 1,            -- 软删除标记
    created_at  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP
                              ON UPDATE CURRENT_TIMESTAMP   -- 自动更新时间
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 任务表
CREATE TABLE tasks (
    id          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    title       VARCHAR(200) NOT NULL,
    description TEXT,                                       -- 可为空的长文本
    status      ENUM('todo', 'doing', 'done') NOT NULL DEFAULT 'todo',
    priority     TINYINT UNSIGNED NOT NULL DEFAULT 2,       -- 1高 2中 3低
    user_id     BIGINT UNSIGNED NOT NULL,                   -- 归属用户
    due_date    DATE,
    created_at  DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    CONSTRAINT fk_task_user                               -- 外键：任务归属用户
        FOREIGN KEY (user_id) REFERENCES users(id),
    INDEX idx_status (status),
    INDEX idx_user_status (user_id, status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 插入：字段名和值一一对应；password_hash 实际存哈希值，此处演示用占位
INSERT INTO users (username, email, password_hash)
VALUES ('xiaoming', 'xiaoming@test.com', 'hashed_password_1'),
       ('xiaohong', 'xiaohong@test.com', 'hashed_password_2');
INSERT INTO tasks (title, description, status, priority, user_id, due_date)
VALUES ('写周报', '整理本周进展', 'todo', 1, 1, '2026-10-01'),
       ('修登录Bug', '验证码不发', 'doing', 1, 1, '2026-09-28'),
       ('需求评审', '新功能评审会', 'todo', 2, 2, '2026-09-30');
-- 查询：* 只用于临时调试，正式代码写明字段
SELECT id, username, email FROM users WHERE is_active = 1;
-- 更新：UPDATE/DELETE 必带 WHERE，否则全表遭殃！
UPDATE tasks SET status = 'done' WHERE id = 1;
-- 删除：企业里常用“软删除”——置 is_active=0 而不是物理删除
UPDATE users SET is_active = 0 WHERE id = 2;

-- 按优先级升序、创建时间倒序；LIMIT 偏移量, 条数
SELECT id, title, status, priority FROM tasks
ORDER BY priority ASC, created_at DESC
LIMIT 0, 10;                        -- 第 1 页，每页 10 条
SELECT DISTINCT status FROM tasks;   -- 有哪些状态值

-- 每个用户、每种状态各有多少任务
SELECT u.username, t.status, COUNT(*) AS cnt
FROM tasks t
JOIN users u ON t.user_id = u.id
GROUP BY u.username, t.status
HAVING cnt >= 1                     -- HAVING 过滤分组结果（WHERE 过滤行）
ORDER BY u.username, cnt DESC;

-- INNER JOIN：只保留两边都能匹配上的行（小红不会出现）
SELECT u.username, t.title
FROM users u INNER JOIN tasks t ON t.user_id = u.id;
-- LEFT JOIN：左表全保留，右表匹配不上的填 NULL（小红 + NULL 出现）
SELECT u.username, t.title
FROM users u LEFT JOIN tasks t ON t.user_id = u.id;
-- RIGHT JOIN：右表全保留（本例结果与 INNER 相近，实践中少用）

-- 查询“没有任何任务”的用户：NOT EXISTS 是最清晰的写法
SELECT id, username FROM users u
WHERE NOT EXISTS (
    SELECT 1 FROM tasks t WHERE t.user_id = u.id
);

-- 建表时已建过 idx_status，这里补充一个组合索引示范
CREATE INDEX idx_due_status ON tasks (due_date, status);
-- EXPLAIN：查看 SQL 如何执行（最重要的一招）
EXPLAIN SELECT * FROM tasks WHERE status = 'todo';

-- 对索引列用函数或运算，索引失效：
SELECT * FROM tasks WHERE YEAR(created_at) = 2026;   -- 全表扫描
SELECT * FROM tasks WHERE id + 1 = 10;               -- 全表扫描
-- 改写为作用于常量的形式，索引恢复生效：
SELECT * FROM tasks
WHERE created_at >= '2026-01-01' AND created_at < '2027-01-01';
SELECT * FROM tasks WHERE id = 9;
-- LIKE 以 % 开头也无法使用索引
SELECT * FROM tasks WHERE title LIKE '%周报%';       -- 全表扫
SELECT * FROM tasks WHERE title LIKE '周报%';        -- 可用索引

START TRANSACTION;
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
UPDATE accounts SET balance = balance + 100 WHERE user_id = 2;
COMMIT;      -- 确认提交，两步同时生效
-- ROLLBACK; -- 出问题时撤销上面所有操作

pip install pymysql cryptography
"""pymysql_demo.py — 原生数据库操作（教学用，生产请用 ORM）"""
import pymysql
# 连接数据库
conn = pymysql.connect(
    host="127.0.0.1",
    port=3306,
    user="root",
    password="root123456",
    database="taskdb",
    charset="utf8mb4",
)
try:
    with conn.cursor() as cursor:
        # 插入：占位符 %s 防止 SQL 注入（值由驱动转义，绝不手拼字符串！）
        cursor.execute(
            "INSERT INTO tasks (title, status, priority, user_id) "
            "VALUES (%s, %s, %s, %s)",
            ("学习 MySQL", "todo", 2, 1),
        )
        new_id = cursor.lastrowid
        print(f"新任务 ID: {new_id}")
        # 查询
        cursor.execute(
            "SELECT id, title, status FROM tasks WHERE status = %s",
            ("todo",),
        )
        rows = cursor.fetchall()
        for row in rows:
            print(row)
    conn.commit()      # 增删改必须 commit 才真正生效
finally:
    conn.close()

pip install "sqlalchemy>=2.0" pymysql cryptography
"""db_engine.py — 创建数据库引擎"""
from sqlalchemy import create_engine
# mysql+pymysql://用户:密码@主机:端口/库名?参数
DATABASE_URL = (
    "mysql+pymysql://root:root123456@127.0.0.1:3306/taskdb"
    "?charset=utf8mb4"
)
# echo=True 会打印执行的 SQL，学习阶段强烈建议开启
engine = create_engine(DATABASE_URL, echo=True, pool_size=5, max_overflow=10)
with engine.connect() as conn:
    result = conn.execute(
        __import__("sqlalchemy").text("SELECT VERSION()")
    )
    print("MySQL 版本:", result.scalar())

"""models.py — 用类描述表结构"""
from datetime import datetime
from sqlalchemy import (
    BigInteger, DateTime, Enum, ForeignKey, Index,
    String, Text, func,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
    """所有模型的基类：SQLAlchemy 2.x 官方推荐写法"""
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(
        BigInteger, primary_key=True, autoincrement=True
    )
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    password_hash: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.current_timestamp()
    )
    # 关系：一个用户有多条任务。backfill 双向引用
    tasks: Mapped[list["Task"]] = relationship(back_populates="owner")
class Task(Base):
    __tablename__ = "tasks"
    __table_args__ = (
        Index("idx_user_status", "user_id", "status"),
    )
    id: Mapped[int] = mapped_column(
        BigInteger, primary_key=True, autoincrement=True
    )
    title: Mapped[str] = mapped_column(String(200))
    description: Mapped[str | None] = mapped_column(Text, default=None)
    status: Mapped[str] = mapped_column(
        Enum("todo", "doing", "done", name="task_status"),
        default="todo",
    )
    priority: Mapped[int] = mapped_column(default=2)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    due_date: Mapped[str | None] = mapped_column()
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.current_timestamp()
    )
    owner: Mapped["User"] = relationship(back_populates="tasks")
if __name__ == "__main__":
    # 一次性建表（已有表则跳过）。生产环境用 Alembic 迁移，见 5.7
    from db_engine import engine
    Base.metadata.create_all(engine)
    print("建表完成")

"""crud_demo.py — ORM 增删改查"""
from sqlalchemy import select
from sqlalchemy.orm import Session
from db_engine import engine
from models import Task, User
# ---------- 增 ----------
with Session(engine) as session:
    user = User(
        username="xiaoming",
        email="xiaoming@test.com",
        password_hash="hashed_xxx",
    )
    session.add(user)
    session.flush()               # 发送 INSERT，拿到自增 id，但不提交
    print(f"新用户 ID: {user.id}")
    task = Task(title="学完 ORM", user_id=user.id, priority=1)
    session.add(task)
    session.commit()              # 事务提交：两个对象一起入库
    print(f"新任务 ID: {task.id}")
# ---------- 查 ----------
with Session(engine) as session:
    # 主键查询
    found = session.get(Task, 1)
    print(f"get: {found.title}")
    # 条件查询：select + where，风格接近原生 SQL
    stmt = (
        select(Task)
        .where(Task.status == "todo")
        .order_by(Task.priority, Task.id.desc())
        .limit(10)
    )
    tasks = session.scalars(stmt).all()      # .all() 取全部
    for t in tasks:
        print(f"{t.id}: {t.title} (优先级 {t.priority})")
    one = session.scalars(
        select(User).where(User.username == "xiaoming")
    ).first()
    print(one.email)
# ---------- 改 ----------
with Session(engine) as session:
    task = session.get(Task, 1)
    task.status = "done"          # 直接改属性
    session.commit()
# ---------- 删 ----------
with Session(engine) as session:
    task = session.get(Task, 2)
    session.delete(task)
    session.commit()

"""relations_demo.py — 关系查询与 N+1 问题"""
from sqlalchemy import select
from sqlalchemy.orm import Session, joinedload, selectinload
from db_engine import engine
from models import Task, User
with Session(engine) as session:
    # 顺藤摸瓜：任务 -> 所属用户（自动 JOIN）
    task = session.get(Task, 1)
    print(f"{task.title} 的主人是 {task.owner.username}")
    # 反向：用户 -> 任务列表
    user = session.get(User, 1)
    print(f"{user.username} 的任务数: {len(user.tasks)}")
    # 查询时预加载，避免 N+1 查询（重点！）
    # 错误示范：循环里访问 t.owner 会触发 N 次额外 SQL
    stmt = select(Task).options(joinedload(Task.owner))
    for t in session.scalars(stmt):
        print(f"{t.title} -> {t.owner.username}")   # 只有 1 条 SQL

"""session_pattern.py — 企业级会话管理模板"""
from sqlalchemy import select
from sqlalchemy.orm import Session
from db_engine import engine
from models import Task
def complete_task(task_id: int) -> Task:
    """把任务标记为完成：异常自动回滚，成功自动提交"""
    with Session(engine) as session:          # Session 用完自动关闭
        with session.begin():                 # 事务块：出错自动 ROLLBACK
            task = session.get(Task, task_id)
            if task is None:
                raise ValueError(f"任务 {task_id} 不存在")
            task.status = "done"
        return task
result = complete_task(1)
print(f"已完成: {result.title}")

pip install alembic
alembic init migrations            # 生成 migrations/ 目录和 alembic.ini

sqlalchemy.url = mysql+pymysql://root:root123456@localhost:3306/taskdb
编辑 migrations/env.py，在文件顶部（target_metadata = None 之前）加入：
import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from models import Base            # 导入你的模型
target_metadata = Base.metadata    # 原本为 None，替换为这一行

# 自动对比模型与数据库差异，生成迁移脚本
alembic revision --autogenerate -m "create users and tasks"
# 执行迁移（建表/改表生效）
alembic upgrade head
# 常用回退命令
alembic downgrade -1        # 回退一个版本
alembic history             # 查看迁移历史

mkdir fastapi_demo && cd fastapi_demo
python -m venv .venv
# 激活虚拟环境后：
pip install "fastapi[standard]"
创建 main.py：
"""main.py — 第一个 FastAPI 应用"""
from fastapi import FastAPI
app = FastAPI(title="我的第一个 API", version="0.1.0")
@app.get("/")
def root():
    return {"message": "Hello, FastAPI!"}
@app.get("/health")
def health_check():
    """健康检查接口：部署时供容器/负载均衡探活"""
    return {"status": "ok"}

fastapi dev main.py
# 旧版本命令：uvicorn main:app --reload

"""params_demo.py — 三种参数一次讲清
启动: fastapi dev params_demo.py（把 app 暴露为 app 即可）
"""
from fastapi import FastAPI
app = FastAPI()
# 1) 路径参数：URL 里的变量 /users/42
@app.get("/users/{user_id}")
def get_user(user_id: int):
    # user_id 自动转成 int；访问 /users/abc 会直接返回 422 校验错误
    return {"user_id": user_id, "name": "小明"}
# 2) 查询参数：URL 问号后面 ?skip=0&limit=10
@app.get("/tasks")
def list_tasks(skip: int = 0, limit: int = 10, keyword: str | None = None):
    return {
        "skip": skip,
        "limit": limit,
        "keyword": keyword,   # 不传就是 None
    }
# 3) 请求体：POST/PUT 放在请求正文的 JSON
from pydantic import BaseModel
class CreateTaskRequest(BaseModel):
    title: str
    priority: int = 2
@app.post("/tasks")
def create_task(req: CreateTaskRequest):
    return {"received": req.model_dump()}

"""pydantic_demo.py — 校验规则
运行: python pydantic_demo.py
"""
from pydantic import BaseModel, Field, EmailStr, field_validator
from enum import Enum
class Status(str, Enum):
    """继承 str 让它可以直接 JSON 序列化"""
    todo = "todo"
    doing = "doing"
    done = "done"
class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    priority: int = Field(default=2, ge=1, le=3)      # ge>=1, le<=3
    status: Status = Status.todo                       # 枚举校验
    owner_email: EmailStr                               # 邮箱格式
    tags: list[str] = []
    description: str | None = None                      # 可选字段
    @field_validator("title")
    @classmethod
    def strip_title(cls, v: str) -> str:
        """自定义校验：去掉首尾空白后不能为空"""
        v = v.strip()
        if not v:
            raise ValueError("标题不能是纯空白")
        return v
# 模拟客户端提交的数据
data = {
    "title": "  学习 Pydantic  ",
    "owner_email": "xiaoming@test.com",
    "tags": ["python", "backend"],
}
task = TaskCreate(**data)
print(task.title)              # "学习 Pydantic"（空白被剥掉）
print(task.model_dump())       # 转字典
# 试试非法数据，看校验如何拦截
try:
    TaskCreate(title="", priority=9, owner_email="不是邮箱")
except Exception as e:
    print(type(e).__name__)    # ValidationError
    for err in e.errors():
        print(f"  字段 {err['loc']}: {err['msg']}")

"""response_model_demo.py — 输出过滤"""
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class UserInDB(BaseModel):
    """数据库层的完整用户"""
    id: int
    username: str
    email: str
    password_hash: str      # 敏感字段
class UserOut(BaseModel):
    """对外暴露的用户"""
    id: int
    username: str
    email: str
fake_db_user = UserInDB(
    id=1, username="xiaoming",
    email="xiaoming@test.com", password_hash="secret_hash",
)
@app.get("/users/me", response_model=UserOut)
def read_me():
    return fake_db_user     # 返回完整对象，但 FastAPI 只输出 UserOut 的字段
# 实际响应: {"id":1,"username":"xiaoming","email":"xiaoming@test.com"}

"""errors_demo.py — HTTPException 与自定义异常"""
from fastapi import FastAPI, HTTPException, status
app = FastAPI()
fake_tasks = {1: "写周报", 2: "修Bug"}
@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    if task_id not in fake_tasks:
        # 抛出 404，detail 会出现在响应 JSON 里
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"任务 {task_id} 不存在",
        )
    return {"id": task_id, "title": fake_tasks[task_id]}
@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    if task_id not in fake_tasks:
        raise HTTPException(status_code=404, detail="任务不存在")
    del fake_tasks[task_id]
    # 204 无内容：删除成功的标准返回
    raise ...  # 占位，见下方说明

@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int):
    if task_id not in fake_tasks:
        raise HTTPException(status_code=404, detail="任务不存在")
    del fake_tasks[task_id]
    return None

pip install python-multipart
"""upload_demo.py — 文件上传"""
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    """UploadFile 比普通 bytes 多了文件名、类型等元信息"""
    contents = await file.read()
    # 保存到本地（注意：文件名要做安全处理，防止路径穿越攻击）
    safe_name = file.filename.replace("/", "_").replace("\\", "_")
    with open(f"uploads/{safe_name}", "wb") as f:
        f.write(contents)
    return {
        "filename": file.filename,
        "size": len(contents),
        "content_type": file.content_type,
    }
提示：大文件用 await file.read() 一次性读入会占内存，流式写法是 while chunk := await file.read(1024 * 1024): f.write(chunk)。

"""di_demo.py — 依赖注入入门"""
from fastapi import Depends, FastAPI, Query
app = FastAPI()
class Pagination:
    """分页参数对象：一处定义，处处复用"""
    def __init__(self, page: int = Query(1, ge=1),
                 size: int = Query(10, ge=1, le=100)):
        self.offset = (page - 1) * size
        self.limit = size
        self.page = page
        self.size = size
@app.get("/tasks")
def list_tasks(pg: Pagination = Depends()):
    # Depends() 括号里不写函数时，自动使用参数的类型 Pagination
    return {
        "page": pg.page,
        "size": pg.size,
        "items": ["任务A", "任务B"][: pg.size],
    }
@app.get("/users")
def list_users(pg: Pagination = Depends()):
    # 第二个接口零成本复用同样的分页能力
    return {"page": pg.page, "size": pg.size, "items": ["小明", "小红"]}

"""auth_chain.py — 嵌套依赖模拟登录"""
from fastapi import Depends, FastAPI, Header, HTTPException
app = FastAPI()
fake_tokens = {"token-alice": 1}      # 模拟 token 存储
fake_users = {1: {"id": 1, "username": "alice", "role": "admin"}}
def parse_token(authorization: str = Header(...)) -> str:
    """第一层：从请求头取 token。Header(...) 声明必填"""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="token 格式错误")
    return authorization.removeprefix("Bearer ")
def get_current_user(token: str = Depends(parse_token)) -> dict:
    """第二层：token 换用户"""
    user_id = fake_tokens.get(token)
    if user_id is None:
        raise HTTPException(status_code=401, detail="token 无效或过期")
    return fake_users[user_id]
@app.get("/me")
def read_me(user: dict = Depends(get_current_user)):
    return user
@app.get("/admin/stats")
def admin_stats(user: dict = Depends(get_current_user)):
    if user["role"] != "admin":
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return {"total_users": 100}

"""middleware_demo.py — 计时中间件与 CORS"""
import time
import uuid
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
@app.middleware("http")
async def add_process_time(request: Request, call_next):
    """给每个响应加耗时头，并生成请求追踪 ID"""
    request_id = str(uuid.uuid4())[:8]
    start = time.perf_counter()
    response = await call_next(request)          # 放行给后续处理
    cost = (time.perf_counter() - start) * 1000
    response.headers["X-Process-Time"] = f"{cost:.2f}ms"
    response.headers["X-Request-ID"] = request_id
    print(f"[{request_id}] {request.method} {request.url.path} {cost:.1f}ms")
    return response
# CORS：前端跨域访问后端的必备配置（本地开发常配 localhost:5173）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],   # 前端地址，生产环境写真实域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.get("/ping")
def ping():
    return {"pong": True}

"""async_demo.py — 异步正确用法"""
import asyncio
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
app = FastAPI()
async def fake_remote_call() -> dict:
    """模拟调外部 API：等待 1 秒，但事件循环期间可服务其他请求"""
    await asyncio.sleep(1)
    return {"data": "来自远方"}
@app.get("/async-route")
async def async_route():
    """规则一：路由内部全是 await 操作时，声明 async def"""
    result = await fake_remote_call()
    return result
@app.get("/sync-route")
def sync_route():
    """规则二：路由里有阻塞调用（如同步数据库驱动）时，声明普通 def
    FastAPI 会自动把它放进线程池，不会卡死事件循环"""
    import time
    time.sleep(1)                    # 模拟同步阻塞
    return {"ok": True}
@app.get("/mixed-route")
async def mixed_route():
    """规则三：async 路由里必须调用阻塞函数时，手动丢进线程池"""
    import time
    result = await run_in_threadpool(time.sleep, 1)
    return {"ok": True}

"""background_demo.py — 后台任务"""
import asyncio
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
async def send_email(to: str, subject: str) -> None:
    await asyncio.sleep(2)                 # 模拟耗时的邮件发送
    print(f"邮件已发送: {to} [{subject}]")
@app.post("/register")
async def register(email: str, bg: BackgroundTasks):
    # 注册流程立刻完成并返回，邮件在后台异步发送
    bg.add_task(send_email, email, "欢迎注册")
    return {"message": "注册成功，欢迎邮件稍后送达"}

"""database.py — 数据库依赖（配合第 5 章 models.py 使用）"""
from collections.abc import Generator
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker
DATABASE_URL = (
    "mysql+pymysql://root:root123456@127.0.0.1:3306/taskdb?charset=utf8mb4"
)
engine = create_engine(DATABASE_URL, pool_size=5, max_overflow=10)
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
def get_db() -> Generator[Session, None, None]:
    """依赖函数：请求开始建会话，结束自动关闭，异常自动回滚"""
    db = SessionLocal()
    try:
        yield db              # 路由执行期间，会话在这里“挂起”
    finally:
        db.close()
"""task_router.py — 路由使用数据库"""
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Session
from database import get_db
from models import Task
app = FastAPI()
class TaskOut(BaseModel):
    id: int
    title: str
    status: str
    model_config = {"from_attributes": True}   # v2 写法：允许从 ORM 对象构造
@app.get("/tasks", response_model=list[TaskOut])
def list_tasks(db: Session = Depends(get_db)):
    stmt = select(Task).order_by(Task.id.desc()).limit(20)
    tasks = db.scalars(stmt).all()
    return tasks                    # TaskOut 自动从 ORM 对象转换
@app.patch("/tasks/{task_id}/done", response_model=TaskOut)
def complete_task(task_id: int, db: Session = Depends(get_db)):
    task = db.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="任务不存在")
    task.status = "done"
    db.commit()
    db.refresh(task)               # 重新加载，拿到最新数据
    return task

"""routers/users.py — 用户域路由"""
from fastapi import APIRouter
router = APIRouter(prefix="/users", tags=["用户管理"])
@router.get("/me")
def read_me():
    return {"username": "alice"}
@router.get("/{user_id}")
def read_user(user_id: int):
    return {"user_id": user_id}
"""main_with_routers.py — 主应用注册路由"""
from fastapi import FastAPI
from database import get_db          # noqa: F401 确保依赖可用
from routers import users
app = FastAPI(title="多模块应用")
app.include_router(users.router)
# 最终路径: /users/me 和 /users/{user_id}

taskflow/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 实例、路由注册、中间件
│   ├── config.py            # 配置管理（pydantic-settings）
│   ├── database.py          # 引擎、SessionLocal、get_db
│   ├── models/              # SQLAlchemy 模型（表结构）
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── task.py
│   ├── schemas/             # Pydantic 模型（接口出入参）
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── task.py
│   │   └── common.py         # 统一响应格式
│   ├── crud/                 # 数据访问层
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── task.py
│   ├── services/            # 业务逻辑层
│   │   ├── __init__.py
│   │   └── task_service.py
│   ├── api/                  # 路由层
│   │   ├── __init__.py
│   │   ├── deps.py           # 公共依赖（get_db、get_current_user）
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── auth.py
│   │       ├── users.py
│   │       └── tasks.py
│   ├── core/                 # 核心工具
│   │   ├── __init__.py
│   │   ├── security.py       # 密码哈希、JWT
│   │   └── exceptions.py    # 业务异常
│   └── utils/
│       └── __init__.py
├── tests/                    # 测试代码（第 11 章）
│   ├── conftest.py
│   └── test_tasks.py
├── migrations/               # Alembic 迁移（第 5 章）
├── .env                      # 环境变量（不进 Git！）
├── .env.example              # 环境变量模板（进 Git）
├── .gitignore
├── alembic.ini
├── pyproject.toml
├── Dockerfile                # 第 12 章
├── docker-compose.yml        # 第 12 章
└── requirements.txt

pip install pydantic-settings
"""app/config.py — 三环境配置管理"""
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
    """应用配置：从环境变量和 .env 文件读取，自动类型校验"""
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,       # 环境变量名不分大小写
    )
    # ---- 应用 ----
    app_name: str = "TaskFlow"
    debug: bool = False
    environment: str = "dev"        # dev / test / prod
    # ---- 数据库 ----
    database_url: str = "mysql+pymysql://root:root123456@127.0.0.1:3306/taskdb"
    # ---- 安全 ----
    secret_key: str = "dev-only-secret-change-me"
    access_token_expire_minutes: int = 30
@lru_cache                          # 缓存：全进程只读一次配置
def get_settings() -> Settings:
    return Settings()

# .env.example —— 复制为 .env 后按环境修改
ENVIRONMENT=dev
DEBUG=true
DATABASE_URL=mysql+pymysql://root:root123456@127.0.0.1:3306/taskdb
SECRET_KEY=please-generate-a-random-one
ACCESS_TOKEN_EXPIRE_MINUTES=30
cp .env.example .env     # 新同事拉代码后的第一步

{"code": 0, "message": "ok", "data": {"id": 1, "title": "写周报"}}
"""app/schemas/common.py — 统一响应封装"""
from typing import Any, Generic, TypeVar
from pydantic import BaseModel
T = TypeVar("T")
class Resp(BaseModel, Generic[T]):
    """标准响应：code=0 成功，非 0 为业务错误码"""
    code: int = 0
    message: str = "ok"
    data: T | None = None
def ok(data: Any = None, message: str = "ok") -> dict:
    return {"code": 0, "message": message, "data": data}
def fail(code: int, message: str) -> dict:
    return {"code": code, "message": message, "data": None}
# 路由中使用：
@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    task = task_service.get_task(task_id)
    return ok(data=task)

"""app/core/exceptions.py — 业务异常"""
class BizError(Exception):
    """业务异常基类：code 对应统一响应的业务错误码"""
    code = 1000
    message = "业务错误"
class TaskNotFoundError(BizError):
    code = 1001
    message = "任务不存在"
class PermissionDeniedError(BizError):
    code = 1003
    message = "无权操作此资源"
"""app/main.py 片段 — 全局异常处理器"""
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from app.core.exceptions import BizError
app = FastAPI()
@app.exception_handler(BizError)
async def biz_error_handler(request: Request, exc: BizError):
    """任何路由抛出业务异常，都会在这里转成统一响应格式"""
    return JSONResponse(
        status_code=400,
        content={"code": exc.code, "message": exc.message, "data": None},
    )
@app.exception_handler(Exception)
async def unhandled_error_handler(request: Request, exc: Exception):
    """兜底：未预期异常返回 500，细节不暴露给客户端（记日志！）"""
    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": "服务器内部错误", "data": None},
    )

pip install "passlib[bcrypt]" "bcrypt==4.0.1"   # 钉住版本：passlib 停更，bcrypt>=4.1 会报 "trapped error reading bcrypt version"
"""app/core/security.py — 密码与令牌工具"""
from datetime import datetime, timedelta, timezone
import jwt   # PyJWT（encode/decode 调用方式与 python-jose 兼容）
from jwt import PyJWTError as JWTError
from passlib.context import CryptContext
# bcrypt 算法（自动加盐，同密码两次哈希结果也不同）
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)
def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)
# ---- 快速体验 ----
if __name__ == "__main__":
    h = hash_password("my-secret-123")
    print("哈希值:", h)
    print("正确密码校验:", verify_password("my-secret-123", h))   # True
    print("错误密码校验:", verify_password("wrong", h))          # False

pip install PyJWT   # 订正：python-jose 维护弱（2024 年算法混淆 CVE），官方文档现行推荐 PyJWT
继续在 app/core/security.py 中添加：
# 追加到 app/core/security.py ----
SECRET_KEY = "change-me-in-env"       # 生产环境必须用强随机值并放 .env
ALGORITHM = "HS256"
def create_access_token(subject: str, expires_minutes: int = 30,
                         extra: dict | None = None) -> str:
    """签发 JWT：subject 通常是 user_id 字符串"""
    payload = {
        "sub": subject,
        "exp": datetime.now(timezone.utc) + timedelta(minutes=expires_minutes),
        "iat": datetime.now(timezone.utc),        # 签发时间
    }
    if extra:
        payload.update(extra)                     # 可塞角色等非敏感信息
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict:
    """验证并解码 JWT：签名不对/过期都会抛 JWTError"""
    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])