05 Python 连接 MySQL 及 Web 开发框架 — 学习指南

学习理念：这是从"写 Python 脚本"到"构建 Web 服务"的跃迁。 FastAPI + SQLAlchemy 是后续所有 AI 项目的后端底座，理解请求-响应周期和 ORM 映射比背 API 更重要。

学习路径

协程基础 → Web 规范 → FastAPI → SQLAlchemy → Python 连接 MySQL

① 🟡 协程基础（async/await 到底在干什么）
② 🟢 WSGI vs ASGI（两种 Web 规范）
③ 🟢 FastAPI 快速入门（路由/参数/请求体）
④ 🟡 SQLAlchemy ORM（对象关系映射）
⑤ 🟢 Python 连接 MySQL（pymysql 基础）

本节 AI 替代率：~93% | 人工干预率：~7%

AI 擅长：写 FastAPI 路由、SQLAlchemy 模型定义、CRUD 代码。
人类需决策：ORM 关系映射设计、异步 vs 同步选型、项目结构组织。

1. 协程基础

李永乐式比喻：你在厨房同时做三件事——
同步：先烧水，干等水烧开了再切菜，切完菜再炒菜（一件一件做）
多线程：请三个厨师各干各的（但厨房就那么大，三个厨师会撞到一起→线程安全）
协程：你一个人，烧水的时候去看书，水开了关火回来继续看书。主动让出 CPU，不需要额外的厨师

async / await 语法

python

import asyncio

# async 定义协程函数
async def fetch_data(url):
    print(f"开始请求: {url}")
    # await 让出 CPU，等待 IO 完成
    await asyncio.sleep(1)  # 模拟网络请求
    print(f"请求完成: {url}")
    return f"data from {url}"

# 运行协程
async def main():
    # 方式一：逐个执行
    result1 = await fetch_data("url1")
    result2 = await fetch_data("url2")  # 这会串行执行
    
    # 方式二：并发执行（这才是协程的价值）
    tasks = [
        fetch_data("url1"),
        fetch_data("url2"),
        fetch_data("url3"),
    ]
    results = await asyncio.gather(*tasks)

# 启动
asyncio.run(main())

核心理解：
async def 声明这个函数是协程，可以暂停/恢复
await 真正执行并等待——遇到 IO 时主动让出，别人先跑
协程适合 IO 密集型（网络请求、文件读写、数据库查询）
协程不适合 CPU 密集型（数学计算、数据处理）

2. WSGI vs ASGI

李永乐式比喻：
WSGI = 单车道隧道。一次只能过一辆车（一个请求）。Python 的传统 Web 框架（Flask/Django）都用这个。
ASGI = 多车道立交桥。可以同时过很多辆车，还能走自行车和行人（WebSocket 等长连接）。FastAPI 用这个。

对比项	WSGI	ASGI
全称	Web Server Gateway Interface	Asynchronous Server Gateway Interface
同步/异步	同步	异步
支持 WebSocket	❌	✅
性能	一般	高并发场景更好
代表框架	Flask, Django（传统模式）	FastAPI, Starlette
服务器	Gunicorn, uWSGI	Uvicorn, Daphne

python

# FastAPI 本身是异步的，但也可以运行同步代码
from fastapi import FastAPI
import time

app = FastAPI()

# 同步路由（FastAPI 会用线程池执行，不阻塞事件循环）
@app.get("/sync")
def read_sync():
    time.sleep(1)
    return {"message": "同步"}

# 异步路由（推荐，IO 密集型场景）
@app.get("/async")
async def read_async():
    await asyncio.sleep(1)
    return {"message": "异步"}

3. FastAPI 快速入门

李永乐式比喻：FastAPI 就像一个智能快递分拣站——
快递员送来包裹（HTTP 请求）
分拣员看地址（URL 路径），送到对应的分拣口（路由函数）
同一个分拣口还可以细分：是特快专递（GET）、是退货（POST）、还是改地址（PUT）

第一个 FastAPI 程序

python

# 安装：pip install fastapi uvicorn

from fastapi import FastAPI

app = FastAPI(title="我的 API")

@app.get("/")
def root():
    return {"message": "Hello World"}

# 运行：uvicorn main:app --reload
# 访问：http://127.0.0.1:8000
# 文档：http://127.0.0.1:8000/docs（自动生成！）

路径参数

python

@app.get("/users/{user_id}")
def get_user(user_id: int):  # 类型注解 → 自动校验
    return {"user_id": user_id, "name": "Alice"}

# 访问 /users/42 → {"user_id": 42, "name": "Alice"}
# 访问 /users/abc → 自动返回 422 校验错误

请求体参数

python

from pydantic import BaseModel

# 定义数据模型（Pydantic）
class UserCreate(BaseModel):
    name: str
    age: int
    email: str | None = None  # 可选字段

@app.post("/users")
def create_user(user: UserCreate):
    # FastAPI 自动解析 JSON 请求体
    # 自动校验字段类型
    # 自动生成 API 文档
    return {
        "name": user.name,
        "age": user.age,
        "email": user.email,
    }

# 请求体：
# POST /users
# {"name": "Alice", "age": 25, "email": "alice@test.com"}

# 响应：
# {"name": "Alice", "age": 25, "email": "alice@test.com"}

路由分发

python

# 项目较大时，把路由拆到多个文件

# user.py
from fastapi import APIRouter

router = APIRouter(prefix="/users", tags=["用户管理"])

@router.get("/")
def list_users():
    return [{"id": 1, "name": "Alice"}]

@router.get("/{user_id}")
def get_user(user_id: int):
    return {"id": user_id, "name": "Alice"}

# main.py
from fastapi import FastAPI
from user import router as user_router

app = FastAPI()
app.include_router(user_router)

FastAPI 自动交互文档

# 启动后自动生成两个文档地址：
http://127.0.0.1:8000/docs          # Swagger UI（推荐，可在线测试 API）
http://127.0.0.1:8000/redoc         # ReDoc（更美观的文档）

# 你不需要额外写文档，FastAPI 从代码类型注解自动生成

4. SQLAlchemy ORM

李永乐式比喻：SQLAlchemy 就像一个翻译官——
你（Python 代码）说："给我找一个叫 Alice 的用户"
翻译官（ORM）把它翻译成："SELECT * FROM user WHERE name='Alice'"
然后翻译官把数据库返回的行数据，打包成 Python 对象还给你

安装

bash

pip install sqlalchemy pymysql

定义模型（创建表）

python

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime
from sqlalchemy.orm import declarative_base
from datetime import datetime

# 创建引擎（连接数据库）
engine = create_engine(
    "mysql+pymysql://root:root@localhost:3306/school",
    echo=True  # 打印 SQL 语句，调试用
)

# 模型基类
Base = declarative_base()

# 定义模型（= 定义表结构）
class Student(Base):
    __tablename__ = "student"  # 表名
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    age = Column(Integer)
    score = Column(Float)
    created = Column(DateTime, default=datetime.now)

# 创建表（首次运行）
Base.metadata.create_all(engine)

增删改查

python

from sqlalchemy.orm import sessionmaker

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# --- 增 ---
new_student = Student(name="张三", age=20, score=85.5)
session.add(new_student)
session.commit()  # 提交事务

# 批量增加
session.add_all([
    Student(name="李四", age=22, score=90),
    Student(name="王五", age=21, score=78),
])
session.commit()

# --- 查 ---
# 查所有
students = session.query(Student).all()

# 条件查询
students = session.query(Student).filter(Student.score > 80).all()

# 查单个
student = session.query(Student).filter(Student.name == "张三").first()

# 模糊查询
students = session.query(Student).filter(Student.name.like("%张%")).all()

# 排序
students = session.query(Student).order_by(Student.score.desc()).all()

# 聚合
from sqlalchemy import func
avg_score = session.query(func.avg(Student.score)).scalar()

# --- 改 ---
student = session.query(Student).filter(Student.name == "张三").first()
if student:
    student.score = 95
    session.commit()

# --- 删 ---
student = session.query(Student).filter(Student.name == "王五").first()
if student:
    session.delete(student)
    session.commit()

# 关闭会话
session.close()

反向生成模型

python

# 如果数据库已经有表，可以自动生成 Python 模型代码
# 命令行：
# sqlacodegen mysql+pymysql://root:root@localhost:3306/school > models.py

# 这会生成 models.py，里面包含了数据库中所有表的 Python 模型定义
# 不需要手动写 Column 定义

5. Python 连接 MySQL（pymysql）

python

# pip install pymysql

import pymysql

# 连接
conn = pymysql.connect(
    host="localhost",
    port=3306,
    user="root",
    password="root",
    database="school",
    charset="utf8mb4"
)

cursor = conn.cursor()

# 查询
cursor.execute("SELECT * FROM student WHERE score > %s", (80,))
results = cursor.fetchall()
for row in results:
    print(row)

# 插入
cursor.execute(
    "INSERT INTO student (name, age, score) VALUES (%s, %s, %s)",
    ("新同学", 20, 88)
)
conn.commit()

cursor.close()
conn.close()

pymysql vs SQLAlchemy 选型：
简单脚本、少量 SQL → pymysql（轻量直接）
项目开发、复杂关系 → SQLAlchemy（ORM 的便利性远超直接写 SQL）

6. 完整示例：FastAPI + SQLAlchemy

python

# 项目结构
# project/
# ├── main.py          # FastAPI 入口
# ├── models.py        # SQLAlchemy 模型
# ├── schemas.py       # Pydantic 模型（请求/响应）
# └── database.py      # 数据库连接

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:root@localhost:3306/school"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# models.py
from sqlalchemy import Column, Integer, String, Float
from database import Base

class Student(Base):
    __tablename__ = "student"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50))
    age = Column(Integer)
    score = Column(Float)

# schemas.py
from pydantic import BaseModel

class StudentCreate(BaseModel):
    name: str
    age: int
    score: float

class StudentResponse(StudentCreate):
    id: int
    class Config:
        from_attributes = True  # 支持 ORM 模式

# main.py
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from database import SessionLocal
from models import Student
from schemas import StudentCreate, StudentResponse

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/students/", response_model=StudentResponse)
def create_student(student: StudentCreate, db: Session = Depends(get_db)):
    db_student = Student(**student.dict())
    db.add(db_student)
    db.commit()
    db.refresh(db_student)
    return db_student

@app.get("/students/", response_model=list[StudentResponse])
def list_students(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return db.query(Student).offset(skip).limit(limit).all()

AI 协作指南

AI 能做的：
  - 写完整的 FastAPI 路由和 Pydantic 模型
  - 生成 SQLAlchemy 模型定义和 CRUD 操作
  - 把数据库表结构翻译成 Python ORM 模型
  - 调试异步代码中的常见错误

人类需决策的：
  - 同步还是异步？大部分场景异步用 async def，同步用 def
  - 项目结构怎么组织？路由拆分到什么粒度
  - ORM 关系映射：一对多、多对多的配置
  - 数据库连接池大小、超时设置

最高效的学习方式：
  - 描述 API 需求给 AI，让 AI 生成 FastAPI 代码
  - 把数据库表结构描述给 AI，让 AI 生成 SQLAlchemy 模型
  - 重点理解请求-响应周期和 ORM 映射原理
  - 不背 FastAPI/SQLAlchemy 的 API，用的时候问 AI

附录：原始资料处理说明

原始文件	处理方式
`FastAPI&SQLAlchemy1.0.docx`	内容已整合到本文档第 3~4 章
`Python连接MySQL.docx`	内容已整合到第 5 章
`MySQLl的卸载与安装.docx`	不单独处理，MySQL 安装见模块 04
视频（17 个）	跳过
代码（demo/routers 目录）	结构已在第 6 章完整示例中展示

05 Python 连接 MySQL 及 Web 开发框架 — 学习指南 ​

学习路径 ​

1. 协程基础 ​

async / await 语法 ​

2. WSGI vs ASGI ​

3. FastAPI 快速入门 ​

第一个 FastAPI 程序 ​

路径参数 ​

请求体参数 ​

路由分发 ​

FastAPI 自动交互文档 ​

4. SQLAlchemy ORM ​

安装 ​

定义模型（创建表） ​

增删改查 ​

反向生成模型 ​

5. Python 连接 MySQL（pymysql） ​

6. 完整示例：FastAPI + SQLAlchemy ​

AI 协作指南 ​

附录：原始资料处理说明 ​

05 Python 连接 MySQL 及 Web 开发框架 — 学习指南

学习路径

1. 协程基础

async / await 语法

2. WSGI vs ASGI

3. FastAPI 快速入门

第一个 FastAPI 程序

路径参数

请求体参数

路由分发

FastAPI 自动交互文档

4. SQLAlchemy ORM

安装

定义模型（创建表）

增删改查

反向生成模型

5. Python 连接 MySQL（pymysql）

6. 完整示例：FastAPI + SQLAlchemy

AI 协作指南

附录：原始资料处理说明