一、为什么要学 SQLAlchemy？

写 Python 操作数据库，新手第一反应是用 pymysql + 原生 SQL，但写着写着就会遇到一堆问题：

  • 😫 SQL 语句拼到吐血，字符串拼接容易出 bug
  • 😫 字段改了，SQL 里漏改一处，排查半天
  • 😫 不同数据库 SQL 语法不一样，换库就得重写
  • 😫 手动拼参数，一不小心就 SQL 注入
  • 😫 查询结果是元组，还得自己转字典/对象

SQLAlchemy 就是来解决这些问题的。 它是 Python 生态最强大的 ORM（对象关系映射）框架，核心思想就是：

用 Python 类定义表，用对象操作数据，不用写原生 SQL。

简单说就是：

# 以前：写 SQL
cursor.execute("SELECT * FROM users WHERE id = 1")
现在：用对象
user = session.execute(select(User).where(User.id == 1)).scalar_one()

代码更优雅、更安全、更易维护，这就是 ORM 的魅力。

二、SQLAlchemy 两大核心体系

SQLAlchemy 分两层，新手先搞清楚这个：

层级 名称 作用 适用场景
底层 Core SQL 表达式语言，用 Python 函数拼 SQL 复杂查询、性能敏感
上层 ORM 对象关系映射，用类和对象操作表 日常 CRUD、业务开发

💡 90% 的业务开发用 ORM 就够了，本文重点讲 ORM，Core 做补充介绍。

核心组件一览

Engine（引擎）
└── 负责数据库连接、SQL 执行
Session（会话）
└── 负责 ORM 操作的上下文，增删改查都通过它
Model（模型）
└── Python 类和数据库表的映射
Query（查询）
└── 链式调用构造查询条件

三、环境准备

3.1 安装

# 基础安装
pip install sqlalchemy
配合 MySQL 使用（需要驱动）
pip install sqlalchemy pymysql
异步版（后面会讲）
pip install sqlalchemy[asyncio] aiomysql

3.2 版本说明

  • SQLAlchemy 1.4.x：过渡版本，同时支持旧 API 和新 API
  • SQLAlchemy 2.0+：当前主流，推荐学习，本文基于 2.0 标准写法

🔥 注意：网上很多老教程是 1.x 写法，session.query() 这种是旧 API，2.0 推荐用 select() 新写法，本文全部用 2.0 标准写法。

四、第一步：连接数据库

4.1 创建 Engine

Engine 是 SQLAlchemy 的入口，负责管理数据库连接池。

from sqlalchemy import create_engine
MySQL 连接
engine = create_engine(
"mysql+pymysql://用户名:密码@localhost:3306/数据库名?charset=utf8mb4",
echo=True,        # 打印执行的 SQL，调试用
pool_size=10,     # 连接池大小
max_overflow=20,  # 最大溢出连接数
pool_recycle=3600 # 连接回收时间（秒）
)
SQLite 连接（本地文件，适合测试，不用装数据库）
engine = create_engine("sqlite:///./test.db")

连接 URL 格式：

数据库类型+驱动://用户名:密码@主机:端口/库名?参数
数据库 URL 示例
MySQL mysql+pymysql://user:pass@localhost/db
PostgreSQL postgresql+psycopg2://user:pass@localhost/db
SQLite sqlite:///./test.db

💡 新手建议先用 SQLite 练习，不用装 MySQL，零环境成本。

4.2 创建 Session

Session 是 ORM 操作的会话，所有增删改查都通过它。

from sqlalchemy.orm import sessionmaker
创建 Session 工厂
SessionLocal = sessionmaker(
bind=engine,
autoflush=False,
autocommit=False,
expire_on_commit=False
)
获取 session
session = SessionLocal()
用完记得关
session.close()

最佳实践：上下文管理器

from contextlib import contextmanager
@contextmanager
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
使用
with get_db() as db:
# 数据库操作
pass

五、第二步：定义模型（建表）

5.1 声明基类

所有模型都要继承这个基类：

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

5.2 定义用户表（新手入门经典示例）

用 Python 类定义一张用户表，类属性就是表字段，就像定义一个普通类一样简单。

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from datetime import datetime

class User(Base):
    """用户表"""
    # 表名（数据库里真实的表名）
    __tablename__ = "users"

    # 主键：用户ID，自增
    id = Column(Integer, primary_key=True, autoincrement=True, comment="用户ID")
    
    # 用户名，不能为空，建索引
    username = Column(String(50), nullable=False, unique=True, index=True, comment="用户名")
    
    # 邮箱
    email = Column(String(100), unique=True, comment="邮箱")
    
    # 密码（实际项目存加密后的）
    password = Column(String(255), nullable=False, comment="密码")
    
    # 年龄
    age = Column(Integer, comment="年龄")
    
    # 是否激活
    is_active = Column(Boolean, default=True, comment="是否激活")
    
    # 创建时间：插入时自动赋值
    created_at = Column(DateTime, default=datetime.now, comment="创建时间")
    
    # 更新时间：每次更新自动刷新
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间")

    # 打印对象时显示的内容，方便调试
    def __repr__(self):
        return f"<User(id={self.id}, username='{username}')>"

5.3 常用字段类型

表格

字段类型 Python 类型 说明
Integer int 整数
String(length) str 字符串，需指定长度
Text str 长文本
Float float 浮点数
Boolean bool 布尔值
DateTime datetime 日期时间
Date date 日期
Enum enum.Enum 枚举
JSON dict/list JSON 字段

5.4 常用列参数

表格

参数 作用
primary_key=True 主键
autoincrement=True 自增
nullable=False 不允许为空
default=xxx 默认值
unique=True 唯一约束（值不能重复）
index=True 建索引，查询更快
comment="xxx" 字段注释
onupdate=xxx 更新时自动赋值

5.5 执行建表

# 创建所有表（继承 Base 的类都会建）
Base.metadata.create_all(engine)

运行完这行，去数据库里看，users 表已经建好了！

⚠️ create_all 只会建不存在的表，已存在的表不会修改结构。改字段要用 Alembic 做迁移（后面会讲）。

六、第三步：增删改查（CRUD）

最核心的部分来了！四大操作：增（Create）、删（Delete）、改（Update）、查（Read）。

6.1 新增数据（Create）

from sqlalchemy import select

# 1. 创建一个 User 对象（就像 new 一个普通类）
user = User(
    username="zhangsan",
    email="zhangsan@example.com",
    password="123456",
    age=25
)

# 2. 添加到 session（相当于暂存）
session.add(user)

# 3. 提交事务（真正写到数据库）
session.commit()

# 4. 刷新（获取自增ID等数据库生成的值）
session.refresh(user)

print(user.id)       # 自增主键，比如 1
print(user.created_at)  # 自动生成的创建时间

批量新增：

# 一次加多条
users = [
    User(username="lisi", email="lisi@example.com", password="123456", age=22),
    User(username="wangwu", email="wangwu@example.com", password="123456", age=30),
    User(username="zhaoliu", email="zhaoliu@example.com", password="123456", age=28),
]

session.add_all(users)
session.commit()

6.2 查询数据（Read）

SQLAlchemy 2.0 推荐用 select() 写法，链式调用，非常灵活。

from sqlalchemy import select, and_, or_

# ---------- 1. 根据主键查询 ----------
user = session.get(User, 1)
print(user.username)  # zhangsan

# ---------- 2. 查询所有用户 ----------
stmt = select(User)
result = session.execute(stmt)
users = result.scalars().all()  # 得到 User 对象列表

for u in users:
    print(u.id, u.username)

# ---------- 3. 条件查询（WHERE）----------
# 查年龄大于25的用户
stmt = select(User).where(User.age > 25)
users = session.execute(stmt).scalars().all()

# ---------- 4. 多条件（AND）----------
# 查年龄>20 且 已激活的用户
stmt = select(User).where(
    User.age > 20,
    User.is_active == True
)
# 或者用 and_（效果一样）
stmt = select(User).where(
    and_(
        User.age > 20,
        User.is_active == True
    )
)

# ---------- 5. OR 条件 ----------
# 查年龄<20 或者 年龄>30 的用户
stmt = select(User).where(
    or_(
        User.age < 20,
        User.age > 30
    )
)

# ---------- 6. 模糊查询（LIKE）----------
# 查用户名包含 "zhang" 的用户
stmt = select(User).where(User.username.like("%zhang%"))

# ---------- 7. 排序（ORDER BY）----------
# 按年龄倒序（从大到小）
stmt = select(User).order_by(User.age.desc())

# 按年龄正序（从小到大）
# stmt = select(User).order_by(User.age.asc())

# ---------- 8. 分页（LIMIT + OFFSET）----------
# 第2页，每页2条
stmt = select(User).offset(2).limit(2)

# ---------- 9. 取第一条 ----------
user = session.execute(stmt).scalars().first()

# ---------- 10. 统计数量 ----------
from sqlalchemy import func
count = session.execute(
    select(func.count(User.id))
).scalar()
print(f"总用户数：{count}")

6.3 更新数据（Update）

# 1. 先查出来
user = session.get(User, 1)

# 2. 直接改属性（就像改普通对象属性一样）
user.age = 26
user.email = "zhangsan_new@example.com"

# 3. 提交
session.commit()

批量更新（适合一次改多条）：

from sqlalchemy import update

# 把所有未激活的用户改成激活
stmt = (
    update(User)
    .where(User.is_active == False)
    .values(is_active=True)
)

session.execute(stmt)
session.commit()

6.4 删除数据（Delete）

# 1. 先查出来
user = session.get(User, 1)

# 2. 删除
session.delete(user)

# 3. 提交
session.commit()

批量删除：

from sqlalchemy import delete

# 删除所有年龄小于18的用户
stmt = delete(User).where(User.age < 18)
session.execute(stmt)
session.commit()

七、第四步：高级查询技巧

7.1 分组聚合

from sqlalchemy import func

# 按年龄分组，统计每个年龄有多少人
stmt = select(
    User.age,
    func.count(User.id).label("user_count")
).group_by(User.age)

result = session.execute(stmt).all()
for row in result:
    print(f"年龄 {row.age}：{row.user_count} 人")

7.2 排序 + 取前 N 个

# 取年龄最大的 3 个用户
stmt = (
    select(User)
    .order_by(User.age.desc())
    .limit(3)
)
top3 = session.execute(stmt).scalars().all()

7.3 IN 查询

# 查 id 在 [1, 3, 5] 里的用户
stmt = select(User).where(User.id.in_([1, 3, 5]))

7.4 动态构造查询（非常实用）

根据前端传的参数，动态加查询条件：

stmt = select(User)

# 如果传了用户名，就加用户名条件
if username:
    stmt = stmt.where(User.username.like(f"%{username}%"))

# 如果传了最小年龄
if min_age:
    stmt = stmt.where(User.age >= min_age)

# 如果传了最大年龄
if max_age:
    stmt = stmt.where(User.age <= max_age)

# 最后统一排序
stmt = stmt.order_by(User.created_at.desc())

users = session.execute(stmt).scalars().all()

八、第五步：关系映射（一对多 / 多对多）

真实项目表和表之间是有关系的，SQLAlchemy 用 relationship 来管理。

8.1 一对多（最常用）

比如：一个用户有多篇文章。

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Article(Base):
    """文章表"""
    __tablename__ = "articles"

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False, comment="标题")
    content = Column(Text, comment="内容")
    
    # 外键：关联用户表的 id
    user_id = Column(Integer, ForeignKey("users.id"), comment="作者ID")

    # 反向关联：通过文章找到作者
    author = relationship("User", back_populates="articles")

class User(Base):
    __tablename__ = "users"
    # ... 其他字段
    
    # 正向关联：一个用户有很多文章
    articles = relationship("Article", back_populates="author", cascade="all, delete-orphan")

使用：

# 查询用户时，自动带出他的所有文章
user = session.get(User, 1)
print(user.articles)  # Article 对象列表

# 通过文章反向找作者
article = session.get(Article, 1)
print(article.author)  # 对应的 User 对象

8.2 多对多

比如：学生和课程，一个学生选多门课，一门课有多个学生。

from sqlalchemy import Table

# 中间表（学生选课表）
student_course = Table(
    "student_course",
    Base.metadata,
    Column("student_id", Integer, ForeignKey("students.id"), primary_key=True),
    Column("course_id", Integer, ForeignKey("courses.id"), primary_key=True)
)

class Student(Base):
    __tablename__ = "students"
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    
    courses = relationship("Course", secondary=student_course, back_populates="students")

class Course(Base):
    __tablename__ = "courses"
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    
    students = relationship("Student", secondary=student_course, back_populates="courses")

九、第六步：异步 SQLAlchemy（FastAPI 标配）

现在 Web 开发都用异步了，SQLAlchemy 2.0 原生支持异步，搭配 FastAPI 绝配。

9.1 安装依赖

pip install sqlalchemy[asyncio] aiomysql

9.2 异步 Engine 和 Session

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

# 异步引擎
engine = create_async_engine(
    "mysql+aiomysql://user:pass@localhost/db?charset=utf8mb4",
    echo=True
)

# 异步 Session 工厂
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

9.3 异步 CRUD

from sqlalchemy import select

async def get_user(user_id: int):
    """根据ID查用户"""
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalars().first()

async def create_user(user: User):
    """新增用户"""
    async with AsyncSessionLocal() as session:
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

async def list_users():
    """查询所有用户"""
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User))
        return result.scalars().all()

9.4 FastAPI 集成

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalars().first()

十、常见踩坑与最佳实践

❌ 坑 1：查询后 commit 了，再访问属性报错

原因：expire_on_commit=True（默认），提交后对象会过期，再访问属性会重新查数据库，但 session 已经关了。 解决：创建 session 时设 expire_on_commit=False

❌ 坑 2：N+1 查询问题

原因：循环里访问关联属性，每次都查一次数据库。比如遍历 10 个用户，每个用户的文章都查一次，总共 1+10=11 次查询。 解决：用 joinedload 或 selectinload 预加载

from sqlalchemy.orm import selectinload

# 一次查询把用户和文章都查出来
stmt = select(User).options(selectinload(User.articles))
users = session.execute(stmt).scalars().all()

❌ 坑 3：忘记 commit

原因：ORM 操作都是在内存里，不 commit 不会写到数据库。 解决：增删改操作后一定要 session.commit()

❌ 坑 4：session 泄漏

原因：用完 session 没关，连接池耗尽。 解决：用上下文管理器或依赖注入，确保 session 关闭。

✅ 最佳实践

  1. 一个请求一个 session，不要全局共用一个 session
  2. 查询用 select () 新写法，别用旧的 query ()
  3. 批量操作用 bulk 方法，比循环 add 快很多
  4. 复杂查询用 Core，简单 CRUD 用 ORM
  5. 生产环境关掉 echo，避免打印大量 SQL 日志

十一、完整项目结构参考

project/
├── db/
│   ├── __init__.py
│   ├── base.py          # Base 基类
│   ├── models.py        # 所有模型
│   └── session.py       # engine 和 session 工厂
├── crud/
│   ├── __init__.py
│   └── user.py          # 用户相关 CRUD
├── schemas/
│   └── user.py          # Pydantic 模型（请求/响应）
└── main.py              # 入口

十二、全文总结

  1. SQLAlchemy 是 Python 最强大的 ORM，分 Core（底层）和 ORM（上层）两层，日常开发用 ORM 就够了。
  2. 四大核心概念：Engine（连接引擎）、Session（会话）、Model（模型）、Query（查询）。
  3. 建表用类定义，字段类型、约束、索引都用 Python 代码描述，不用写原生 DDL。
  4. CRUD 全是对象操作，增用 add、查用 select、改直接赋值、删用 delete，最后 commit 提交。
  5. 关系映射：一对多用 relationship + 外键，多对多加中间表，关联查询记得预加载防 N+1。
  6. 异步版本：用 create_async_engine + AsyncSession，完美适配 FastAPI。
  7. 注意事项：记得 commit、记得关 session、注意 N+1 问题、生产关 echo。

觉得有用的话，点赞 + 收藏 + 关注，持续更新 Python 后端开发全套实战教程！

# python
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

2026最新AI Agent高频面试题（分类型精简详解）

本文梳理AI Agent高频面试考点，覆盖基础概念、架构、推理规划、工具调用、记忆、RAG融合、多智能体与工程落地。区分Agent、Workflow与普通工具调用，讲解ReAct、Plan‑and‑Execute等主流范式，分析三级记忆机制与上下文溢出的解决办法，对比传统RAG与Agent‑RAG差异，介绍多智能体协作模式。同时总结落地痛点、优化手段和安全风控要点，划分初/中/高级面试复习重点，可

avatar AI Agent技术社区
cover

深入理解 AI Agent · AGENT #03：从单 Agent 到多 Agent

avatar AI Agent技术社区
cover

AI Agent安全风险启示：工业巡检机器人的AI安全管控

avatar AI Agent技术社区