版本: 3.1.0 Python 后端能力

Python 数据访问：PyMySQL 与 SQLAlchemy 入门

集中整理 Python 项目里的数据访问主线：先用 PyMySQL 理解连接与事务，再用 SQLAlchemy 建立 ORM 认知。

这篇文章只回答一件事：Python 程序该如何与关系型数据库打交道。

一、先区分两个角色

MySQL：关系型数据库，负责结构化数据持久化
ORM：对象关系映射，让你用对象方式操作数据库，而不是频繁手写 SQL

可以先记住一句话：

先会手写 SQL，你才知道 ORM 在帮你抽象什么
学会 ORM，你才知道大型项目为什么不愿到处散落 SQL

二、使用 PyMySQL 连接 MySQL

pymysql 到今天依然是常见方案之一。

import pymysql

conn = pymysql.connect(
    host="127.0.0.1",
    port=3306,
    user="root",
    password="your-password",
    database="demo",
    charset="utf8mb4",
)

try:
    with conn.cursor() as cursor:
        cursor.execute(
            "INSERT INTO user (name, password) VALUES (%s, %s)",
            ("alex", "123456"),
        )
    conn.commit()
except Exception:
    conn.rollback()
    raise
finally:
    conn.close()

查询示例

import pymysql

conn = pymysql.connect(
    host="127.0.0.1",
    port=3306,
    user="root",
    password="your-password",
    database="demo",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
)

try:
    with conn.cursor() as cursor:
        cursor.execute("SELECT id, name FROM user")
        rows = cursor.fetchall()
        for row in rows:
            print(row)
finally:
    conn.close()

先记住这几个原则

写操作后记得 commit()
写操作如果中途失败，记得 rollback()
优先使用参数化 SQL
连接和游标都要及时关闭
查询结果常见方法是 fetchone()、fetchmany()、fetchall()

三、为什么还要学习 ORM

当业务逐渐复杂时，手写 SQL 会开始暴露这些问题：

表结构和代码模型之间缺少统一映射
常见增删改查重复度高
关系查询表达不一致
大量 SQL 分散在不同模块里，维护成本上升

这时 ORM 的价值就出来了。

常见选择可以简单理解成：

独立 Python 项目：常见选择是 SQLAlchemy
Django 项目：通常直接使用 Django ORM

四、SQLAlchemy 的最小认知

`Classic` 映射和 `Declarative` 映射怎么理解

旧版 SQLAlchemy 常见 Classic 映射：

先单独描述表
再把 Python 类映射到表

例如：

from sqlalchemy import Column, Integer, MetaData, String, Table

metadata = MetaData()
user_table = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
    Column("fullname", String(50)),
)

这有助于理解 ORM 的本质，但在今天的新项目里，更推荐声明式写法。

更推荐的 `DeclarativeBase`

定义模型、创建引擎、开启会话三步走：

from sqlalchemy import String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


# 1. 定义模型基类：所有模型都继承它
class Base(DeclarativeBase):
    pass


# 2. 定义模型类：一个类对应一张表
class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(32))
    password: Mapped[str] = mapped_column(String(64))


# 3. 创建引擎：连接字符串格式为 "数据库类型+驱动://用户:密码@主机/库名?参数"
engine = create_engine(
    "mysql+pymysql://root:password@127.0.0.1/demo?charset=utf8mb4",
    echo=True,  # 打印底层 SQL，便于调试学习
)

Mapped[int] 这种写法是 2.0 风格，类型注解同时决定字段类型，IDE 也能给出提示。

五、增删改查速查

下面统一以上面的 User 模型为例。所有写操作都要在 Session 上下文里完成，并通过 commit() 提交事务。

5.1 新增（Create）

from sqlalchemy.orm import Session

with Session(engine) as session:
    # 单条新增
    user = User(name="alex", password="123456")
    session.add(user)        # 加入会话
    session.commit()         # 提交后才会真正写入数据库
    print(user.id)           # 提交后主键会自动回填到对象上

    # 批量新增
    session.add_all([
        User(name="bob", password="bbb"),
        User(name="cathy", password="ccc"),
    ])
    session.commit()

commit() 之后，自增主键 user.id 会被自动填充，无需再次查询。

5.2 查询（Read）

SQLAlchemy 2.0 推荐 select() 写法，语义更清晰。

from sqlalchemy import select, asc, desc, func

with Session(engine) as session:
    # 查全部
    users = session.execute(select(User)).scalars().all()

    # 按主键查一条
    user = session.get(User, 1)          # 等价于 WHERE id=1，最快捷

    # 条件过滤
    users = session.execute(
        select(User).where(User.name == "alex")
    ).scalars().all()

    # 多条件组合：且 / 或
    from sqlalchemy import and_, or_
    users = session.execute(
        select(User).where(
            and_(User.name.like("a%"), User.id > 10)
        )
    ).scalars().all()

    # 排序与分页
    users = session.execute(
        select(User)
        .order_by(desc(User.id))     # 按 id 倒序
        .limit(10)                    # 取前 10 条
        .offset(20)                   # 跳过 20 条（第 3 页）
    ).scalars().all()

    # 只查指定列：返回的是 Row，不是模型对象
    rows = session.execute(
        select(User.id, User.name)
    ).all()
    for r in rows:
        print(r.id, r.name)

    # 聚合查询：count / max / avg
    total = session.execute(
        select(func.count(User.id))
    ).scalar()                         # scalar() 取单值

常用过滤方法速查：

方法	等价 SQL	示例
`==` / `!=`	`=` / `<>`	`User.name == "alex"`
`>` / `<` / `>=` / `<=`	比较符	`User.id > 10`
`.like()`	`LIKE`	`User.name.like("a%")`
`.in_()`	`IN`	`User.name.in_(["alex", "bob"])`
`.is_(None)`	`IS NULL`	`User.name.is_(None)`
`.between()`	`BETWEEN`	`User.id.between(1, 100)`
`and_` / `or_` / `not_`	逻辑符	`and_(cond1, cond2)`

5.3 修改（Update）

修改有两种方式。

方式一：查出来改属性（适合需要先读后写的场景）

with Session(engine) as session:
    user = session.get(User, 1)
    user.password = "new-pwd"          # 直接改对象属性
    session.commit()                   # 提交时自动生成 UPDATE

方式二：直接构造 UPDATE 语句（适合批量更新，不必先查）

from sqlalchemy import update

with Session(engine) as session:
    # 把所有 name='alex' 的用户密码改成 'new-pwd'
    result = session.execute(
        update(User)
        .where(User.name == "alex")
        .values(password="new-pwd")
    )
    session.commit()
    print(result.rowcount)             # 受影响行数

方式二跳过了"先查后改"，数据库只走一次 UPDATE，批量场景下效率更高。

5.4 删除（Delete）

同样两种方式。

方式一：查出来删对象

with Session(engine) as session:
    user = session.get(User, 1)
    session.delete(user)               # 标记删除
    session.commit()                   # 提交时生成 DELETE

方式二：直接构造 DELETE 语句

from sqlalchemy import delete

with Session(engine) as session:
    result = session.execute(
        delete(User).where(User.name == "alex")
    )
    session.commit()
    print(result.rowcount)

5.5 事务与回滚

Session 本身就是一个事务边界。出错时调用 rollback() 回滚：

from sqlalchemy.exc import IntegrityError

with Session(engine) as session:
    try:
        session.add(User(name="alex", password="x"))   # 假设 name 有唯一约束
        session.add(User(name="alex", password="y"))   # 重复，会触发唯一键冲突
        session.commit()
    except IntegrityError:
        session.rollback()             # 回滚整个事务，前面的插入也作废
        print("插入失败，已回滚")

退出 with 块时若未 commit()，会自动 rollback()，不会脏写数据库。

六、一对多关系快速上手

真实业务里表与表之间往往有关联。以"部门 - 员工"为例：

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship


class Department(Base):
    __tablename__ = "department"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(32))

    # 反向关系：通过 department.employees 直接拿到该部门所有员工
    employees: Mapped[list["Employee"]] = relationship(back_populates="department")


class Employee(Base):
    __tablename__ = "employee"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(32))
    dept_id: Mapped[int] = mapped_column(ForeignKey("department.id"))

    # 正向关系：通过 employee.department 直接拿到所属部门
    department: Mapped["Department"] = relationship(back_populates="employees")

带关系的增查：

with Session(engine) as session:
    # 新增：连同关联对象一起插入
    dept = Department(name="技术部")
    dept.employees = [Employee(name="alex"), Employee(name="bob")]
    session.add(dept)
    session.commit()

    # 查询：关系属性会按需加载（默认懒加载）
    dept = session.get(Department, 1)
    for emp in dept.employees:          # 访问时才发出第二条 SQL
        print(emp.name)

七、一句话速记

MySQL 存数据，PyMySQL 执行 SQL，ORM 把操作组织得更可维护
写操作必 commit()，出错必 rollback()
查询用 select()，过滤用 where()，scalars().all() 取对象列表
改、删都有"对象方式"和"语句方式"两条路，批量场景选语句方式