用奶茶店学会 FastAPI —— 零基础全套教程

# 变量就是贴了标签的杯子，用来装数据
# 就像奶茶杯上写"珍珠奶茶"，杯子里装的就是珍珠奶茶

name = "小明"         # 杯子上贴"name"，装的是文字"小明"
age = 25             # 杯子上贴"age"，装的是数字 25
price = 15.5         # 杯子上贴"price"，装的是小数 15.5
is_vip = True        # 杯子上贴"is_vip"，装的是"是/否" True

# 就像奶茶店的排队单 —— 按顺序排，可以插队、可以划掉
orders = ["小明-珍珠奶茶", "小红-芋泥波波", "小刚-柠檬绿茶"]

# ---------- 查 ----------
orders[0]          # "小明-珍珠奶茶"     （查第1个，索引从0开始）
orders[-1]         # "小刚-柠檬绿茶"      （查最后1个）
orders[0:2]        # ["小明-珍珠奶茶", "小红-芋泥波波"]  （切片：第1到第2个）
len(orders)        # 3                   （查长度：有几单）
"小明-珍珠奶茶" in orders  # True        （查是否存在）

# ---------- 增 ----------
orders.append("小丽-杨枝甘露")            # 追加到末尾
orders.insert(1, "小王-四季春")           # 插队到第2个位置

# ---------- 改 ----------
orders[0] = "小明-双拼奶茶"              # 修改第1个

# ---------- 删 ----------
orders.remove("小红-芋泥波波")            # 按内容删除
deleted = orders.pop()                   # 弹出最后一个（取走+删除）
deleted = orders.pop(0)                  # 弹出第1个
del orders[1]                            # 按位置删除
orders.clear()                           # 清空全部

# ---------- 遍历 ----------
for order in orders:                     # 一个一个处理
    print(f"制作：{order}")

# 就像配方卡 —— 每种原料对应一个用量，凭名称快速查
recipe = {
    "红茶": "200ml",
    "鲜奶": "100ml",
    "珍珠": "30g",
    "糖浆": "15ml"
}

# ---------- 查 ----------
recipe["红茶"]              # "200ml"        （查红茶用量，key 不存在会报错！）
recipe.get("椰果")           # None           （安全查，不存在返回 None）
recipe.get("椰果", "没有")   # "没有"         （不存在返回默认值）
"珍珠" in recipe            # True           （查 key 是否存在）
recipe.keys()               # dict_keys(['红茶','鲜奶','珍珠','糖浆'])  （所有原料名）
recipe.values()             # dict_values(['200ml','100ml','30g','15ml'])  （所有用量）
recipe.items()              # 键值对列表      （遍历用）

# ---------- 增 ----------
recipe["椰果"] = "20g"                      # 添加新原料

# ---------- 改 ----------
recipe["糖浆"] = "10ml"                     # 修改用量（少糖）

# ---------- 删 ----------
del recipe["珍珠"]                          # 删掉珍珠（去珍珠）
removed = recipe.pop("糖浆")                # 取出糖浆并删除
recipe.clear()                              # 清空配方

# ---------- 遍历 ----------
for name, amount in recipe.items():         # 同时拿原料名和用量
    print(f"加{name}：{amount}")

# 就像已经封口的奶茶 —— 不能往里面加料，不能换杯子
fixed_menu = ("珍珠奶茶", "芋泥波波", "柠檬绿茶")  # 注意：用圆括号

# ---------- 查（和列表一样） ----------
fixed_menu[0]        # "珍珠奶茶"
fixed_menu[1:3]      # ("芋泥波波", "柠檬绿茶")
len(fixed_menu)      # 3
"珍珠奶茶" in fixed_menu  # True

# ---------- 增/改/删 ----------
# ❌ 全部不行！元组不可变
# fixed_menu.append("杨枝甘露")   # 报错！
# fixed_menu[0] = "双拼奶茶"      # 报错！

# ---------- 为什么需要元组？ ----------
# 1. 函数返回多个值
def get_position():
    return (120, 35)           # 返回坐标，不应该被修改
x, y = get_position()          # 解包：x=120, y=35

# 2. 字典的 key 必须是不可变类型
order_count = {("珍珠奶茶", "大杯"): 15}  # 元组可以作为 key，列表不行

# 就像「今日售罄」黑板 —— 写了就不重复写，顺序无所谓
sold_out = {"芋泥", "椰果", "布丁"}

# ---------- 查 ----------
"芋泥" in sold_out      # True           （查是否售罄）
len(sold_out)           # 3              （几种售罄）

# ---------- 增 ----------
sold_out.add("红豆")                     # 添加售罄品
sold_out.add("芋泥")                     # 已存在，不会重复添加！

# ---------- 改（没有改，只有删了再加） ----------

# ---------- 删 ----------
sold_out.remove("椰果")                  # 删除（不存在会报错）
sold_out.discard("椰果")                 # 安全删除（不存在也不报错）
item = sold_out.pop()                    # 随机弹出一个

# ---------- 集合特有运算 ----------
a = {"珍珠", "椰果", "布丁"}
b = {"椰果", "布丁", "红豆"}

a | b   # 并集：{"珍珠","椰果","布丁","红豆"}  （所有原料）
a & b   # 交集：{"椰果","布丁"}               （两边都有的）
a - b   # 差集：{"珍珠"}                      （a有b没有的）

# ---------- 去重妙用 ----------
orders = ["珍珠", "芋泥", "珍珠", "芋泥", "柠檬"]
unique = list(set(orders))  # ["珍珠", "芋泥", "柠檬"]  （瞬间去重！）

# 函数就是一套固定的操作流程，给原料 → 出成品
# 就像「做珍珠奶茶」这个工序：输入(茶、奶、珍珠) → 输出(珍珠奶茶)

def make_milk_tea(tea, milk, pearl):      # def = 定义工序
    """制作珍珠奶茶"""
    cup = tea + milk + pearl              # 把原料混在一起
    return cup                            # return = 交出成品


# 调用函数 = 下单
my_drink = make_milk_tea("红茶", "鲜奶", "珍珠")
print(my_drink)  # 输出：红茶鲜奶珍珠

# 类就是模板，用来批量创建相同结构的东西
# 就像奶茶店的「配方卡」—— 规定了每款饮品必须包含哪些信息

class MilkTea:                           # 定义配方模板
    def __init__(self, name, price, size):   # __init__ = 制作时必填的信息
        self.name = name                 # self.name = 这杯奶茶的名字
        self.price = price               # self.price = 这杯奶茶的价格
        self.size = size                 # self.size = 这杯奶茶的规格


# 用模板创建具体的奶茶
tea1 = MilkTea("珍珠奶茶", 15, "大杯")    # 按模板制作第1杯
tea2 = MilkTea("芋泥波波", 18, "中杯")    # 按模板制作第2杯

print(tea1.name)   # 珍珠奶茶
print(tea2.price)  # 18

# 类里不仅能存数据，还能放函数（方法）
# 就像配方卡上除了原料清单，还印了制作步骤

class MilkTea:
    def __init__(self, name, price, size):
        self.name = name
        self.price = price
        self.size = size

    def make_milk_tea(self, tea, milk, pearl):
        """制作工序：把原料混在一起"""
        return f"{self.size} {self.name} = {tea} + {milk} + {pearl}"
        #        ↑ self.size 是这杯茶的规格    ↑ self.name 是这杯茶的名字


# 创建对象 + 调用方法
tea = MilkTea("珍珠奶茶", 15, "大杯")
result = tea.make_milk_tea("红茶", "鲜奶", "珍珠")
print(result)  # 大杯 珍珠奶茶 = 红茶 + 鲜奶 + 珍珠

# Python 3.6+ 支持给变量贴类型标签
# 就像原料罐上贴「白糖」「珍珠」，一看就知道里面是什么

def make_order(name: str, quantity: int, price: float) -> str:
    """
    name: str      → name 应该是文字类型
    quantity: int  → quantity 应该是整数
    price: float   → price 应该是小数
    -> str         → 这个函数返回文字类型
    """
    total = quantity * price
    return f"{name} x{quantity}杯，共{total}元"


# FastAPI 的核心理念就是靠这个类型注解来实现自动化！

# async = 异步，就像奶茶店店员同时处理多个订单
# 普通函数：做完一杯再做下一杯（同步，排队等）
# async 函数：煮茶的时候去收银，收完银茶也煮好了（异步，不空等）

import asyncio

async def boil_water():           # async def = 这是一个可以"不空等"的操作
    await asyncio.sleep(3)        # await = "我先去干别的，3秒后回来"
    return "水烧好了"

async def take_order():
    await asyncio.sleep(1)        # "点单需要1秒，我先去烧水"
    return "订单已记录"

# FastAPI 默认支持 async，让服务器能同时处理几百个请求

-- 创建数据库 = 租一个仓库
CREATE DATABASE milk_tea_shop;

-- 使用这个仓库
USE milk_tea_shop;

-- 创建表 = 在仓库里搭货架
CREATE TABLE orders (
    id INT AUTO_INCREMENT PRIMARY KEY,   -- 订单编号，自动递增，唯一标识
    customer_name VARCHAR(50) NOT NULL,  -- 客户名，最长50字，必填
    tea_name VARCHAR(50) NOT NULL,       -- 饮品名，必填
    quantity INT DEFAULT 1,              -- 数量，默认1杯
    price DECIMAL(10, 2),                -- 价格，最多10位数，2位小数
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP  -- 下单时间，自动填
);

-- 增 (Create) = 新订单记录
INSERT INTO orders (customer_name, tea_name, quantity, price)
VALUES ('小明', '珍珠奶茶', 2, 15.00);

-- 查 (Read) = 查看订单
SELECT * FROM orders WHERE customer_name = '小明';  -- 查小明的所有订单
SELECT * FROM orders ORDER BY created_at DESC;      -- 按时间倒序查看

-- 改 (Update) = 修改订单
UPDATE orders SET quantity = 3 WHERE id = 1;        -- 把1号订单改为3杯

-- 删 (Delete) = 取消订单
DELETE FROM orders WHERE id = 1;                    -- 删除1号订单

"""
SQLAlchemy ORM 完整示例 —— 用 Python 类操作 MySQL
就像用收银机操作仓库，不需要自己写 SQL 搬货架
"""
from sqlalchemy import create_engine, Column, Integer, String, Numeric, DateTime, func
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# ① 连接数据库 = 打开仓库门
#    格式：数据库类型+驱动://用户名:密码@地址:端口/数据库名
engine = create_engine("mysql+pymysql://root:password@localhost:3306/milk_tea_shop", echo=True)

# ② 创建会话工厂
SessionLocal = sessionmaker(bind=engine)

# ③ 基类
class Base(DeclarativeBase):
    pass

# ④ 定义模型 = 设计货架结构
class Order(Base):
    __tablename__ = "orders"           # 对应哪张表

    id = Column(Integer, primary_key=True, autoincrement=True)
    customer_name = Column(String(50), nullable=False)
    tea_name = Column(String(50), nullable=False)
    quantity = Column(Integer, default=1)
    price = Column(Numeric(10, 2))
    created_at = Column(DateTime, server_default=func.now())

# ⑤ 创建表（只在第一次运行时执行）
Base.metadata.create_all(engine)

# ⑥ CRUD 操作
db = SessionLocal()

# ---- 增 ----
new_order = Order(customer_name="小明", tea_name="珍珠奶茶", quantity=2, price=15.00)
db.add(new_order)
db.commit()
db.refresh(new_order)       # 获取数据库生成的自增 id
print(f"新订单编号：{new_order.id}")

# ---- 查 ----
# 查全部
all_orders = db.query(Order).all()
# 按条件查
ming_orders = db.query(Order).filter(Order.customer_name == "小明").all()
# 按主键查
order1 = db.get(Order, 1)

# ---- 改 ----
order1.quantity = 3         # 直接改属性
db.commit()                 # 提交就生效

# ---- 删 ----
db.delete(order1)
db.commit()

db.close()

# 1. 安装 Python（去 python.org 下载安装）
# 检查是否安装成功
python --version   # 应显示 Python 3.10+ 

# 2. 安装 FastAPI 全家桶
pip install "fastapi[standard]"
# 这条命令会装：fastapi + uvicorn(服务器) + pydantic(数据校验) + ...

# 3. 安装 MySQL 连接工具
pip install sqlalchemy pymysql

# 4. 安装数据库驱动（如果需要）
pip install cryptography

# main.py —— 你的第一个 FastAPI 文件

from fastapi import FastAPI

app = FastAPI()  # 创建应用 = 开业！


@app.get("/")    # 定义路由 = 菜单上的第一项
async def root():
    return {"message": "奶茶店开门了！🧋"}


@app.get("/menu")
async def get_menu():
    return {
        "menu": [
            {"name": "珍珠奶茶", "price": 15},
            {"name": "芋泥波波", "price": 18},
            {"name": "柠檬绿茶", "price": 12},
        ]
    }

# 在终端运行（和 main.py 同目录）
fastapi dev main.py

# 看到这个就成功了：
# INFO:     Uvicorn running on http://127.0.0.1:8000

顾客说："我要一杯大杯珍珠奶茶，少糖，加椰果"
         │
         ▼
    ┌─────────┐
    │ 收银台   │  ← 接收订单（接收 HTTP 请求）
    │ 记录需求  │
    └────┬────┘
         │
    ┌────▼────┐
    │ 配方卡   │  ← 校验订单是否合法（Pydantic 校验）
    │ 检查：有  │     "少糖"是合法选项吗？有珍珠吗？
    │ 没有珍珠？│
    └────┬────┘
         │
    ┌────▼────┐
    │ 制作台   │  ← 处理业务逻辑（你的代码）
    │ 按配方做  │     调茶、加料、封口
    └────┬────┘
         │
    ┌────▼────┐
    │ 出餐口   │  ← 返回结果（返回 JSON 响应）
    │ 喊号取餐  │     "38号，你的珍珠奶茶好了！"
    └─────────┘

from fastapi import FastAPI

app = FastAPI()


@app.get("/order/{tea_id}")
async def get_order(tea_id: int):
    """
    为什么贴 :int 标签？
    
    顾客点单：/order/3   → tea_id = 3（数字）  ✅ 正确
    顾客点单：/order/abc → FastAPI 自动报错    ❌ "abc不是数字"
    
    如果没有 :int，tea_id 默认是字符串 "3"，
    你每次都要手动写 tea_id = int(tea_id) 转换，
    还要自己判断能不能转。贴个标签就全自动了！
    """
    menu = {1: "珍珠奶茶", 2: "芋泥波波", 3: "柠檬绿茶"}
    return {"tea_id": tea_id, "name": menu.get(tea_id, "未知")}

from pydantic import BaseModel


# 配方卡模板 —— 规定了定制奶茶需要填什么
class CustomTea(BaseModel):
    tea_name: str                    # 茶底：必填（没有默认值）
    size: str = "中杯"              # 规格：选填，默认中杯
    sugar: str = "全糖"             # 甜度：选填，默认全糖
    toppings: list[str] = []        # 加料：选填，默认不加


@app.put("/order/{order_id}")
async def update_order(
    order_id: int,            # ← 在 URL 路径 { } 里 → 路径参数（订单号）
    tea: CustomTea,           # ← 是 Pydantic 模型 → 请求体 JSON（配方详情）
    note: str | None = None   # ← 普通类型不在路径 → 查询参数 ?note=xxx（备注）
):
    """
    一个请求三种参数，FastAPI 全自动区分：

    顾客发送：
    PUT /order/38?note=少冰
    Body: {"tea_name": "乌龙茶", "size": "大杯", "sugar": "半糖", "toppings": ["珍珠", "椰果"]}

    FastAPI 自动解析为：
    - order_id = 38                                      （从网址提取）
    - tea = CustomTea(tea_name="乌龙茶", size="大杯",    （从请求体 JSON 解析）
                      sugar="半糖", toppings=["珍珠","椰果"])
    - note = "少冰"                                       （从 ?note= 提取）

    你一行解析代码都不用写！
    """
    return {
        "order_id": order_id,
        "tea": tea,
        "note": note,
        "message": f"{order_id}号订单已更新"
    }

from pydantic import BaseModel, Field


class OrderItem(BaseModel):
    """
    为什么每个字段都要声明类型？
    因为奶茶店不能接受"来一杯随便"这种订单。
    必须明确：什么茶、什么规格、什么甜度。
    Pydantic 就是帮你强制执行这些规则的。
    """
    tea_name: str = Field(min_length=1, max_length=20)
    # ↑ 茶底名：1-20个字，"来一杯"通过，空字符串不通过

    quantity: int = Field(ge=1, le=99)
    # ↑ 数量：1-99杯，ge=greater or equal, le=less or equal
    #   点 0 杯不通过，点 100 杯也不通过（怕恶搞）

    sugar_level: str = Field(pattern="^(无糖|三分糖|半糖|七分糖|全糖)$")
    # ↑ 甜度：只能是这 5 种之一，写"超级甜"不通过


@app.post("/order/")
async def create_order(order: OrderItem):
    """
    顾客发送：{"tea_name": "", "quantity": 0, "sugar_level": "超级甜"}
    
    FastAPI 自动返回 422 错误，详细告诉你：
    - tea_name: 长度至少为 1
    - quantity: 不能小于 1
    - sugar_level: 只能是 无糖/三分糖/半糖/七分糖/全糖
    
    所有校验自动完成，你只管处理合法订单！
    """
    total = order.quantity * 15  # 假设每杯 15 元
    return {
        "message": f"下单成功：{order.tea_name} x{order.quantity}杯",
        "total": total
    }

from typing import Annotated
from fastapi import Depends


# 步骤1：把公共操作写成函数（相当于把常用配料提前备好）
async def get_current_shop_status():
    """
    为什么单独写这个函数？
    因为「查店铺状态」这个操作在 50 个接口里都需要。
    不写依赖注入的话，每个接口都要写：
        status = check_shop_open()
        if not status: return error
    重复 50 次！
    """
    # 实际项目中这里会查数据库
    return {"is_open": True, "waiting_count": 5}


# 步骤2：定义"需要管理员"的依赖（它又依赖"店铺状态"）
async def check_is_manager(
    status: Annotated[dict, Depends(get_current_shop_status)]
):
    """
    为什么这里又套了一层 Depends？
    因为「查权限」之前必须先知道「店铺开了没」。
    
    依赖链：check_is_manager → get_current_shop_status
    就像：加奶盖 → 需要先有奶茶底
    FastAPI 会自动按顺序调用！
    """
    if not status["is_open"]:
        raise Exception("店铺已关门")
    # 这里还可以继续校验是不是管理员...
    return True


# 步骤3：封装成类型别名，方便到处用
ShopStatusDep = Annotated[dict, Depends(get_current_shop_status)]


@app.get("/menu/")
async def get_menu(status: ShopStatusDep):
    """
    看！只需要写 status: ShopStatusDep
    FastAPI 自动：
    1. 调用 get_current_shop_status()
    2. 把返回的 {"is_open": True, "waiting_count": 5} 赋值给 status
    
    对比传统写法（你需要手动做）：
    def get_menu():
        status = get_current_shop_status()  # 手动调用
        ...
    """
    if not status["is_open"]:
        return {"message": "店铺休息中，菜单暂不可用"}
    return {"menu": [...], "waiting": status["waiting_count"]}


@app.get("/admin/report/")
async def admin_report(is_manager: Annotated[bool, Depends(check_is_manager)]):
    """
    这里形成了两层依赖链：
    请求 → get_current_shop_status() → check_is_manager() → 赋值给 is_manager
    
    全部自动完成！你只写业务逻辑。
    """
    return {"report": "今日销售额：¥3,280"}

import time
from fastapi import Request


@app.middleware("http")
async def shop_middleware(request: Request, call_next):
    """
    为什么叫「中间件」？
    因为它在「顾客进店」和「顾客离店」的中间位置。

    流程（洋葱模型）：
    ┌──────────────────────────────┐
    │  🚪 顾客推门进来              │
    │  ┌────────────────────────┐  │
    │  │ 测温 + 记录到店时间     │  │  ← 中间件前置
    │  │ ┌──────────────────┐   │  │
    │  │ │ 点单/取餐(业务逻辑)│   │  │  ← 你的代码
    │  │ └──────────────────┘   │  │
    │  │ 记录离店时间 + 满意度   │  │  ← 中间件后置
    │  └────────────────────────┘  │
    │  🚪 顾客离开                 │
    └──────────────────────────────┘
    """
    # === 进店时 ===
    start_time = time.perf_counter()
    print(f"🔔 顾客进店: {request.method} {request.url}")

    # === 把请求交给业务处理（点单）===
    response = await call_next(request)

    # === 离店时 ===
    elapsed = time.perf_counter() - start_time
    response.headers["X-Service-Time"] = str(elapsed)
    print(f"✅ 服务完成，耗时 {elapsed:.2f}秒")

    return response

from fastapi import HTTPException
from fastapi.responses import JSONResponse


# 自定义异常：原料不足
class OutOfStock(Exception):
    def __init__(self, item: str, needed: int, available: int):
        self.item = item
        self.needed = needed
        self.available = available


# 注册处理器：原料不足时的应急预案
@app.exception_handler(OutOfStock)
async def out_of_stock_handler(request, exc: OutOfStock):
    """
    为什么单独注册处理器？
    因为「原料不足」这个异常可能在 20 个接口里都会发生。
    与其在每个接口里写 if 判断，不如集中写一个处理器。
    
    就像消防喷淋 —— 装一个系统，整栋楼哪个房间着火都能自动喷水。
    """
    return JSONResponse(
        status_code=503,  # 503 = 服务暂不可用
        content={
            "error": "原料不足，暂时无法制作",
            "detail": f"{exc.item}需要{exc.needed}份，库存仅剩{exc.available}份",
            "suggestion": "请稍后再来或更换其他饮品"
        }
    )


@app.post("/order/bulk/")
async def bulk_order(tea_name: str, quantity: int):
    # 模拟库存检查
    stock = {"珍珠": 10, "椰果": 3, "芋泥": 0}

    if tea_name in stock and stock[tea_name] < quantity:
        # 触发异常 → 自动调用上面的处理器！
        raise OutOfStock(tea_name, quantity, stock[tea_name])

    return {"message": f"{tea_name} x{quantity}杯，马上制作！"}

from fastapi import BackgroundTasks


def print_receipt(order_id: int, tea_name: str, quantity: int):
    """
    为什么这是普通函数而不是 async？
    因为打印小票可能要 3 秒（连打印机、排版、出纸），
    如果让顾客等着小票打印完才收到「下单成功」，
    体验很差。所以丢到后台慢慢打。
    """
    import time
    time.sleep(3)  # 模拟打印耗时
    print(f"🧾 小票已打印：{order_id}号 {tea_name}x{quantity}")


@app.post("/order/quick/")
async def quick_order(
    tea_name: str,
    quantity: int,
    background_tasks: BackgroundTasks  # ← 声明后台任务管理器
):
    """
    时间线：
    0.0s  顾客下单
    0.1s  系统记录订单
    0.2s  返回 {"message": "下单成功！"}  ← 顾客看到这个就走了
    ...    后台：print_receipt() 慢慢打印小票（3秒）
    3.2s  小票打印完成
    
    顾客只需等 0.2 秒，不用等打印的 3 秒！
    """
    # 记录订单...
    order_id = 1024

    # 把打印小票丢到后台
    background_tasks.add_task(print_receipt, order_id, tea_name, quantity)

    # 立刻返回
    return {"message": f"下单成功！{order_id}号订单，请稍候取餐"}

import jwt
from datetime import datetime, timedelta, timezone
from fastapi.security import OAuth2PasswordBearer

SECRET_KEY = "milk-tea-shop-secret-2024"  # 会员卡防伪印章（实际项目用复杂随机字符串）
ALGORITHM = "HS256"

# 告诉系统：会员卡从 /login 接口领取
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")


# 模拟会员数据库
members = {
    "xiaoming": {"password": "123456", "level": "金牌会员", "points": 520},
    "xiaohong": {"password": "111111", "level": "银牌会员", "points": 100},
}


@app.post("/login")
async def login(username: str, password: str):
    """
    为什么返回 token 而不是直接返回用户信息？
    
    就像奶茶店会员卡：
    - 第一次：出示身份证 + 填表 → 领会员卡（登录）
    - 之后每次：刷会员卡就行 → 不用再掏身份证（用 token）
    
    JWT 就是电子会员卡，里面加密存储了你是谁、有效期、
    还有防伪签名（防止伪造会员卡）。
    """
    user = members.get(username)
    if not user or user["password"] != password:
        raise HTTPException(status_code=401, detail="账号或密码错误")

    # 制作会员卡（JWT 令牌）
    token = jwt.encode(
        {
            "sub": username,                                    # 持卡人
            "level": user["level"],                             # 会员等级
            "exp": datetime.now(timezone.utc) + timedelta(hours=2)  # 2小时后过期
        },
        SECRET_KEY,
        algorithm=ALGORITHM
    )

    return {"access_token": token, "token_type": "bearer"}
    # 固定格式：access_token + token_type= bearer
    # 这是 OAuth2 标准，Swagger 文档会自动识别


@app.get("/my-points/")
async def my_points(token: str = Depends(oauth2_scheme)):
    """
    为什么用 Depends(oauth2_scheme)？
    它会自动从请求头 Authorization: Bearer <token> 中提取会员卡号。
    你不用手动解析 HTTP 头！
    
    顾客请求时：
    Header: Authorization: Bearer eyJhbGciOi...
    
    oauth2_scheme 自动提取 eyJhbGciOi... → 赋值给 token
    """
    try:
        # 刷卡验证（解码 JWT）
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        return {
            "username": username,
            "level": payload.get("level"),
            "points": members[username]["points"]
        }
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="会员卡已过期，请重新登录")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="无效的会员卡")

# ❌ 错误写法：动态路由写前面
@app.get("/order/{order_id}")     # {order_id} 会匹配任何内容
async def get_order(order_id: int):
    return {"order": order_id}

@app.get("/order/today-special")  # 永远匹配不到！
async def today_special():        # 因为 /order/today-special 被上面吞了
    return {"special": "买一送一"}  # order_id 变成了 "today-special"


# ✅ 正确写法：固定路由写前面
@app.get("/order/today-special")  # 先匹配 /order/today-special
async def today_special():
    return {"special": "今日特价：珍珠奶茶买一送一！"}

@app.get("/order/{order_id}")     # 再匹配 /order/123 这样的
async def get_order(order_id: int):
    orders = {123: "珍珠奶茶", 456: "芋泥波波"}
    return {"order_id": order_id, "tea": orders.get(order_id, "未知")}

                    顾客进门（请求到达）
                         │
              ┌──────────▼──────────┐
              │   中间件（测温登记）  │  规律五：所有请求必经
              └──────────┬──────────┘
                         │
              ┌──────────▼──────────┐
              │  依赖注入（自动备料） │  规律四：需要什么自动给
              └──────────┬──────────┘
                         │
              ┌──────────▼──────────┐
              │  Pydantic 校验       │  规律三：配方卡校验订单
              │  （检查订单合不合法） │
              └──────────┬──────────┘
                         │
              ┌──────────▼──────────┐
              │  参数自动识别         │  规律一+二：自动分清
              │  （哪是茶名哪是备注） │  路径参数/请求体/查询参数
              └──────────┬──────────┘
                         │
              ┌──────────▼──────────┐
              │  你的业务逻辑         │  你做奶茶
              └──────────┬──────────┘
                         │
              ┌──────────▼──────────┐
              │  异常处理（应急预案） │  规律六：出事自动处理
              └──────────┬──────────┘
                         │
              ┌──────────▼──────────┐
              │  后台任务（外卖配送） │  规律七：耗时操作不等待
              └──────────┬──────────┘
                         │
                    顾客离店（响应返回）

  [1] 写代码           [2] 本地开发         [3] 测试           [4] 部署上线
  ┌──────────┐      ┌──────────┐      ┌──────────┐      ┌──────────┐
  │ main.py  │  →   │ fastapi  │  →   │ 接口测试  │  →   │ 服务器   │
  │ models   │      │ dev      │      │ 单元测试  │      │ 运行     │
  │ routers  │      │ 热重载    │      │          │      │         │
  │ services │      │ /docs    │      │          │      │         │
  └──────────┘      └──────────┘      └──────────┘      └──────────┘

milk_tea_shop/                  # 项目根目录
├── main.py                     # 入口：创建 app，注册路由
├── requirements.txt            # 依赖清单
├── .env                        # 环境变量（密钥、数据库地址等，不要提交 git！）
├── app/                        # 应用代码
│   ├── __init__.py
│   ├── models/                 # 数据模型（Pydantic + SQLAlchemy）
│   │   ├── __init__.py
│   │   ├── schemas.py          # Pydantic 请求/响应模型
│   │   └── database.py         # SQLAlchemy 数据库模型
│   ├── routers/                # 路由（按业务模块拆分）
│   │   ├── __init__.py
│   │   ├── orders.py           # 订单相关接口
│   │   ├── menu.py             # 菜单相关接口
│   │   └── users.py            # 用户相关接口
│   ├── services/               # 业务逻辑层
│   │   ├── __init__.py
│   │   ├── order_service.py    # 订单业务逻辑
│   │   └── user_service.py     # 用户业务逻辑
│   ├── dependencies/           # 依赖注入
│   │   └── auth.py             # 认证、权限依赖
│   └── utils/                  # 工具函数
│       └── response.py         # 统一响应格式
├── tests/                      # 测试
│   ├── test_orders.py
│   └── test_users.py
└── alembic/                    # 数据库迁移（可选）

# 直接用 uvicorn 跑，适合小项目
uvicorn main:app --host 0.0.0.0 --port 8000

# 安装
pip install gunicorn

# 启动 4 个工作进程
gunicorn main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000

# ===== Dockerfile文件 =====
# 就像奶茶店加盟手册，定义了新店从零搭建的每一步

# ① 选择基础镜像：相当于选定店铺装修模板（python:3.11-slim 是精简版，体积小）
FROM python:3.11-slim

# ② 设置工作目录：相当于划定店铺的操作间区域，所有操作都在这个目录里进行
WORKDIR /app

# ③ 先复制依赖文件：相当于先把食材采购清单拿过来（先复制 requirements.txt 再安装，可以利用 Docker 缓存层加速构建）
COPY requirements.txt .

# ④ 安装依赖：相当于按采购清单去进货（--no-cache-dir 不保留安装包缓存，让镜像更小）
RUN pip install --no-cache-dir -r requirements.txt

# ⑤ 复制全部代码：相当于把奶茶配方、操作流程、收银系统全部搬进店里
COPY . .

# ⑥ 启动命令：相当于开店营业，启动服务器接待顾客
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# 构建镜像
docker build -t milk-tea-shop .

# 运行容器
docker run -d -p 8000:8000 milk-tea-shop

# 不要把密码写死在代码里！用环境变量
import os
from dotenv import load_dotenv

load_dotenv()  # 从 .env 文件加载

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")
SECRET_KEY = os.getenv("SECRET_KEY", "dev-secret")

# .env 文件（不要提交到 git！）
DATABASE_URL=mysql+pymysql://root:password@localhost:3306/milk_tea_shop
SECRET_KEY=my-super-secret-key-2024

# 生成依赖清单（项目根目录执行）
pip freeze > requirements.txt

# 新环境一键安装
pip install -r requirements.txt

# requirements.txt 示例
fastapi[standard]==0.115.*
sqlalchemy==2.0.*
pymysql==1.1.*
python-jose[cryptography]==3.3.*
python-dotenv==1.0.*
gunicorn==21.*

                    互联网
                      │
                ┌─────▼─────┐
                │  Nginx    │  反向代理 + HTTPS + 静态文件
                │  :443     │
                └─────┬─────┘
                      │
          ┌───────────┼───────────┐
          │           │           │
    ┌─────▼─────┐ ┌───▼───┐ ┌───▼───┐
    │ Uvicorn   │ │Uvicorn│ │Uvicorn│  4 个 worker 进程
    │ Worker 1  │ │Worker2│ │Worker3│
    └─────┬─────┘ └───┬───┘ └───┬───┘
          │           │           │
          └───────────┼───────────┘
                      │
                ┌─────▼─────┐
                │  MySQL    │  数据库
                └───────────┘

OutOfStock              = 火灾（异常类，定义"发生了什么"）
out_of_stock_handler    = 消防喷淋（处理函数，定义"怎么应对"）
@app.exception_handler  = 接线盒（把"火灾类型"和"喷淋系统"绑定）

# 步骤1：定义异常类（定义"火灾"是什么）
#        就像定义：火灾 = 有火焰 + 有烟雾 + 有温度
class OutOfStock(Exception):             # OutOfStock 是类名（警报器型号）
    def __init__(self, item: str, needed: int, available: int):
        self.item = item                 # 缺什么
        self.needed = needed             # 需要多少
        self.available = available       # 还剩多少


# 步骤2：注册处理器（安装消防喷淋，并告诉它"只扑灭 OutOfStock 类型的火灾"）
@app.exception_handler(OutOfStock)       # 把"这个型号的警报器"和"这个喷淋"绑定
async def out_of_stock_handler(request, exc: OutOfStock):
    #                                  ↑ exc 就是被抛出的 OutOfStock 实例
    #                                  可以访问 exc.item, exc.needed, exc.available
    return JSONResponse(status_code=503, content={"error": f"{exc.item}缺货"})


# 步骤3：业务代码中触发（着火了！拉警报！）
@app.post("/order/")
async def create_order(tea_name: str, quantity: int):
    if stock[tea_name] < quantity:
        raise OutOfStock(tea_name, quantity, stock[tea_name])
        #     ↑ 创建一个 OutOfStock 实例（拉响警报）
        #     FastAPI 自动找到上面注册的 out_of_stock_handler 来"灭火"

from pydantic import BaseModel
from typing import Any


# 定义统一的响应格式
class ApiResponse(BaseModel):
    status: int = 200           # HTTP 状态码
    message: str = "success"    # 提示信息
    data: Any = None            # 实际数据


@app.get("/menu/", response_model=ApiResponse)
async def get_menu():
    """成功时"""
    return ApiResponse(
        status=200,
        message="success",
        data={"menu": ["珍珠奶茶", "芋泥波波"]}
    )
    # 返回：{"status": 200, "message": "success", "data": {"menu": [...]}}


@app.get("/order/{order_id}", response_model=ApiResponse)
async def get_order(order_id: int):
    """失败时"""
    orders = {1: "珍珠奶茶"}
    if order_id not in orders:
        return ApiResponse(
            status=404,
            message="订单不存在",
            data=None
        )
        # 返回：{"status": 404, "message": "订单不存在", "data": null}

    return ApiResponse(
        status=200,
        message="success",
        data={"order_id": order_id, "tea": orders[order_id]}
    )

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
from typing import Any


class ApiResponse(BaseModel):
    status: int = 200
    message: str = "success"
    data: Any = None


app = FastAPI()


# ===== 核心：中间件自动包装正常响应 =====
@app.middleware("http")
async def wrap_response(request: Request, call_next):
    """
    所有正常返回的响应，自动包装成 {status, message, data} 格式。
    
    工作原理：
    1. 先执行你的业务代码，拿到原始响应
    2. 把原始响应内容放到 data 字段里
    3. 包上 status 和 message
    """
    response = await call_next(request)

    # 只包装 JSON 响应（跳过静态文件、HTML 等）
    content_type = response.headers.get("content-type", "")
    if "application/json" in content_type:
        # 读取原始响应体（正确方式：用 Starlette 提供的 body 属性）
        import json
        body_bytes = b""                            # b"" 是空字节串，bytes 类型
        async for chunk in response.body_iterator:  # ✅ 用公开属性 body_iterator
            body_bytes += chunk                     # ✅ 拼接到字节串上

        original_data = json.loads(body_bytes) if body_bytes else None

        # 包装成统一格式
        wrapped = ApiResponse(status=200, message="success", data=original_data)

        return JSONResponse(
            content=jsonable_encoder(wrapped),
            status_code=200,
        )

    return response


# ===== 异常处理：异常也统一格式 =====
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    """HTTP 异常统一包装"""
    return JSONResponse(
        status_code=exc.status_code,  # 保持原始状态码
        content=jsonable_encoder(
            ApiResponse(
                status=exc.status_code,
                message=str(exc.detail),
                data=None
            )
        )
    )


@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    """未知异常统一包装（兜底）"""
    return JSONResponse(
        status_code=500,
        content=jsonable_encoder(
            ApiResponse(status=500, message="服务器内部错误", data=None)
        )
    )


# ===== 业务代码：完全不用管包装 =====
@app.get("/menu/")
async def get_menu():
    """直接返回数据就行，中间件自动包装！"""
    return {"items": ["珍珠奶茶", "芋泥波波", "柠檬绿茶"]}
    # 客户端收到：
    # {"status": 200, "message": "success", "data": {"items": ["珍珠奶茶", "芋泥波波", "柠檬绿茶"]}}


@app.get("/order/{order_id}")
async def get_order(order_id: int):
    """异常也自动包装！"""
    orders = {1: "珍珠奶茶", 2: "芋泥波波"}
    if order_id not in orders:
        raise HTTPException(status_code=404, detail="订单不存在")
        # 客户端收到：
        # {"status": 404, "message": "订单不存在", "data": null}
    return {"order_id": order_id, "tea": orders[order_id]}
    # 客户端收到：
    # {"status": 200, "message": "success", "data": {"order_id": 1, "tea": "珍珠奶茶"}}

class MilkTea:               → 后厨师傅的"工作流程"（内部逻辑用）
class MilkTea(BaseModel):    → 顾客填的"电子点单屏"（接口输入用）

# ===== 方式一：普通类（Python 原生，不继承任何东西）=====
class MilkTea:
    def __init__(self, name, price, size):   # 字段在 __init__ 里手动定义
        self.name = name
        self.price = price
        self.size = size

# 创建对象
tea1 = MilkTea("珍珠奶茶", 15, "大杯")
print(tea1.name)   # 珍珠奶茶
print(tea1.price)  # 15


# ===== 方式二：同名类，但继承 BaseModel =====
from pydantic import BaseModel

class MilkTea(BaseModel):             # ← 只多了 (BaseModel)，写法全变了！
    name: str                         # 字段直接声明，带类型
    price: float                      # 自动校验必须是小数
    size: str = "中杯"                # 可以有默认值
    toppings: list[str] = []          # 复杂的嵌套类型也能声明

# 创建对象（必须用 字段名=值 的方式）
tea2 = MilkTea(name="芋泥波波", price=18, size="大杯")
print(tea2.name)   # 芋泥波波

# 还能从 JSON 直接转！
tea3 = MilkTea(**{"name": "柠檬绿茶", "price": 12})
print(tea3.size)   # 中杯（用了默认值）

# ✅ 用普通类：内部业务逻辑
#    后厨自己用的对象，不需要给顾客看，不需要自动校验
class OrderCalculator:
    def __init__(self):
        self.total = 0
        self.discount = 0

    def apply_coupon(self, code: str):
        if code == "VIP2024":
            self.discount = 0.8
        return self.total * self.discount

calculator = OrderCalculator()
calculator.total = 100
print(calculator.apply_coupon("VIP2024"))  # 80.0


# ✅ 用 Pydantic BaseModel：接口的请求/响应
#    顾客填的点单卡，必须校验格式，必须生成文档
class OrderRequest(BaseModel):
    tea_name: str
    quantity: int = Field(ge=1, le=99)
    coupon_code: str | None = None


@app.post("/order")
async def create_order(order: OrderRequest):
    # FastAPI 自动：校验 quantity ≥ 1、tea_name 非空、生成文档
    # 你拿到的一定是合法数据！
    return {"message": f"{order.tea_name} x{order.quantity} 杯"}