SQLAlchemy 学习与使用

什么是 SQLAlchemy

英 /ˈælkəmi/ —— “艾尔-克-米”，原意是炼金术。

SQLAlchemy 是 Python 中最流行的ORM（Object-Relational Mapping，对象关系映射）框架之一，它将 Python 对象与数据库表进行映射，允许开发者使用 Python 语法操作数据库，无需直接编写原生 SQL 语句（也支持原生 SQL 混合使用），大幅提升数据库操作的简洁性、可读性和可维护性。

核心优势：

• 跨数据库兼容性：支持 MySQL、PostgreSQL、SQLite、Oracle 等主流数据库
• 灵活的查询方式
• 强大的事务支持
• 完善的 ORM 映射机制

适用于从小型项目到大型企业级应用的各类场景。

SQLAlchemy 核心概念

SQLAlchemy 的核心分为两大模块：Core（核心组件）和ORM（对象关系映射）。

• Core是底层的数据库交互引擎，提供 SQL 表达式构建、连接管理等功能；
• ORM是在 Core 之上的封装，实现对象与数据库表的映射，是我们学习的重点。

Core 核心组件

1. Engine（引擎）：SQLAlchemy 的核心入口，负责管理数据库连接，通过连接字符串创建，决定了数据库类型、地址、账号密码等信息。
2. Session（会话）：用于与数据库进行交互的"桥梁"，所有数据库操作（增删改查）都通过 Session 完成，相当于一个临时的数据库连接会话。
3. Base（基类）：所有 ORM 模型类的父类，通过 Base 类创建的子类会自动映射为数据库中的表。
4. Model（模型）：Python 中的类，对应数据库中的一张表，类的属性对应表的字段，类的实例对应表的一行数据。
5. Column（字段）：用于定义模型类的属性（即数据库表的字段），可指定字段类型、主键、非空、默认值等约束。

ORM 对象关系映射

ORM（Object-Relational Mapping，对象关系映射）是一种编程技术，用于实现面向对象编程语言中的对象与关系型数据库中的表之间的映射。

核心优势：

• 无需编写原生 SQL，通过操作 Python 对象即可完成数据库操作，降低学习成本
• 实现数据模型与数据库表的解耦，便于代码维护和迁移（如从 MySQL 切换到 PostgreSQL）
• 自动处理数据类型转换（如 Python 的datetime与数据库的DATETIME）
• 支持事务管理、关联查询等复杂数据库操作

快速开始

1. 环境准备

安装相应的库。首先安装 SQLAlchemy 及对应数据库的驱动（以 MySQL 为例，其他数据库驱动见后续说明）：

# 安装 SQLAlchemy 核心库 pip install sqlalchemy # 安装 MySQL 驱动（常用两种，二选一） pip install pymysql # 纯 Python 实现，兼容性好 pip install mysql-connector-python # MySQL 官方驱动

其他数据库驱动安装：

• PostgreSQL：pip install psycopg2-binary
• SQLite：无需额外安装（Python 自带）
• Oracle：pip install cx_Oracle

安装完成后，在 Python 环境中执行以下代码，无报错则说明安装成功：

importsqlalchemyimportpymysqlprint(sqlalchemy.__version__)# 打印 SQLAlchemy 版本号，如 2.0.49print(pymysql.__version__)# 打印 pymysql 版本号，如 2.2.8

2. 数据库准备

在开始使用 SQLAlchemy 操作 MySQL 前，需提前准备好 MySQL 数据库（手动创建，后续 SQLAlchemy 仅操作表，不创建数据库）：

CREATEDATABASEsqlalchemy_demoCHARACTERSETutf8mb4;

3. 创建 Engine（连接数据库）

首先通过连接字符串创建 Engine，连接字符串的格式根据数据库类型不同而不同：

fromsqlalchemyimportcreate_engine# MySQL + pymysqlengine=create_engine("mysql+pymysql://root:123456@localhost:3306/sqlalchemy_demo?charset=utf8mb4",echo=True,# 开启 SQL 打印，调试时可清晰看到 SQLAlchemy 生成的原生 SQLpool_pre_ping=True,# 每次取连接前先探活，避免拿到失效连接)# 其他数据库的连接字符串示例：# SQLite: create_engine("sqlite:///./demo.db")# PostgreSQL: create_engine("postgresql+psycopg2://user:pass@localhost:5432/demo")

说明：

• root：数据库用户名，替换为自己的数据库账号
• 123456：数据库密码，替换为自己的密码
• sqlalchemy_demo：要连接的数据库名（需提前在 MySQL 中创建）
• echo=True：开启 SQL 打印，调试时可清晰看到 SQLAlchemy 生成的原生 SQL，便于排查问题

4. 创建 Base 基类和 Model 模型

通过 Base 基类创建模型类，映射到数据库中的表（以"用户表 user"为例）：

fromsqlalchemy.ormimportdeclarative_basefromsqlalchemyimportColumn,Integer,String,DateTime,Float,Boolean Base=declarative_base()classUser(Base):__tablename__="user"# 数据库中的表名id=Column(Integer,primary_key=True,autoincrement=True,comment="用户ID")username=Column(String(50),unique=True,nullable=False,comment="用户名")email=Column(String(100),nullable=False,comment="邮箱")age=Column(Integer,comment="年龄")created_at=Column(DateTime,server_default="now()",comment="创建时间")def__repr__(self):returnf"<User(id={self.id}, username={self.username})>"

字段类型说明（常用）：

• Integer：整数类型（对应 MySQL 的INT）
• String(n)：字符串类型，n 为最大长度（对应 MySQL 的VARCHAR(n)）
• DateTime：时间类型（对应 MySQL 的DATETIME）
• Float：浮点数类型（对应 MySQL 的FLOAT/DOUBLE）
• Boolean：布尔类型（对应 MySQL 的TINYINT(1)）

字段约束说明（常用）：

• primary_key=True：设为主键
• autoincrement=True：自增（仅整数主键可用）
• nullable=False：非空约束
• unique=True：唯一约束
• default：默认值
• comment：字段注释

Column常用属性表：

5. 创建数据表

通过 Base 类的create_all()方法，自动根据模型类创建数据库表（如果表已存在，则不会重复创建）：

Base.metadata.create_all(engine)

执行后，会在sqlalchemy_demo数据库中创建user表，可通过 MySQL 客户端查看表结构，验证创建结果。

6. 创建 Session，实现数据库操作

Session 是与数据库交互的核心，所有操作都需通过 Session 完成，步骤为：创建 Session → 执行操作 → 提交事务 → 关闭 Session。

fromsqlalchemy.ormimportSession session=Session(engine)

7. 插入数据

# 单条插入user=User(username="alice",email="alice@example.com",age=20)session.add(user)session.commit()print(user.id)# 提交后主键会回填# 批量插入session.add_all([User(username="bob",email="bob@example.com",age=25),User(username="carol",email="carol@example.com",age=30),])session.commit()

8. 查询数据

# 查询全部users=session.query(User).all()# 查询第一条first=session.query(User).first()# 按主键查询one=session.get(User,1)# 推荐写法# 兼容写法：session.query(User).get(1)

快速开始练习

到这一步，你已经跑通了"建表 → 插入 → 查询"的最小闭环。下面我们把它组织成工程化的结构，正式进入 CRUD 实战。

ORM 核心操作 CRUD

抽取通用配置

将配置抽象书写在sqlalchemy_config.py中统一定义与使用：

# sqlalchemy_config.pyfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportdeclarative_base,sessionmaker DB_URI="mysql+pymysql://root:123456@localhost:3306/sqlalchemy_demo?charset=utf8mb4"engine=create_engine(DB_URI,echo=True,pool_pre_ping=True)Base=declarative_base()SessionLocal=sessionmaker(bind=engine)# 后续用 SessionLocal() 创建会话

模型类

# models.pyfromsqlalchemyimportColumn,Integer,String,DateTimefromsqlalchemy_configimportBaseclassUser(Base):__tablename__="user"id=Column(Integer,primary_key=True,autoincrement=True,comment="用户ID")username=Column(String(50),unique=True,nullable=False,comment="用户名")email=Column(String(100),nullable=False,comment="邮箱")age=Column(Integer,comment="年龄")def__repr__(self):returnf"<User(id={self.id}, username={self.username})>"

CRUD 操作

Read（查询数据）

SQLAlchemy 提供了丰富的查询方法，核心是query()函数，结合过滤条件、排序、分页等操作，适配 MySQL 的查询语法。

基础查询：

session=SessionLocal()# 查询全部all_users=session.query(User).all()# 查询第一条first_user=session.query(User).first()# 按主键查询user=session.get(User,1)

条件查询：

核心是filter()函数添加查询条件，支持 MySQL 所有常用运算符：

# 相等session.query(User).filter(User.username=="alice").first()# 大于 / 小于session.query(User).filter(User.age>18).all()session.query(User).filter(User.age<60).all()# 多条件 AND（逗号分隔即为 AND）session.query(User).filter(User.age>=18,User.age<=30).all()# OR / AND / NOTfromsqlalchemyimportor_,and_,not_ session.query(User).filter(or_(User.age<18,User.age>60)).all()session.query(User).filter(and_(User.age>=18,User.age<=30)).all()# 模糊查询session.query(User).filter(User.username.like("%li%")).all()# INsession.query(User).filter(User.age.in_([20,25,30])).all()

排序、分页查询：

# 排序（asc 升序 / desc 降序）session.query(User).order_by(User.age.desc()).all()session.query(User).order_by(User.id.asc()).all()# 分页：limit 取 N 条，offset 跳过前 M 条session.query(User).order_by(User.id).offset(10).limit(5).all()

分组查询：

fromsqlalchemyimportfunc# 按年龄分组统计人数session.query(User.age,func.count(User.id)).group_by(User.age).all()

常用聚合函数：

Create（新增数据）

新增数据的步骤：创建模型实例 → 将实例添加到会话 → 提交会话（commit），新增后的数据会直接写入 MySQL 数据库。

user=User(username="dave",email="dave@example.com",age=28)session.add(user)session.commit()print(user.id)# 提交后主键自动回填

Delete（删除数据）

删除数据的步骤：查询数据 → 删除实例 → 提交会话，删除操作会直接删除 MySQL 中的对应数据。

user=session.query(User).filter(User.username=="dave").first()ifuser:session.delete(user)session.commit()

Update（更新数据）

更新数据的步骤：查询数据 → 修改实例属性 → 提交会话，更新操作会直接同步到 MySQL 数据库。

user=session.query(User).filter(User.username=="alice").first()ifuser:user.age=21user.email="alice_new@example.com"session.commit()# 修改属性后提交即可，无需 add

关联关系（一对多）

实际业务中，表和表之间往往存在关联。以"一个用户拥有多篇文章"为例：

fromsqlalchemyimportColumn,Integer,String,ForeignKeyfromsqlalchemy.ormimportrelationshipfromsqlalchemy_configimportBaseclassUser(Base):__tablename__="user"id=Column(Integer,primary_key=True,autoincrement=True)username=Column(String(50),unique=True,nullable=False)# 通过 relationship 直接访问该用户的文章articles=relationship("Article",back_populates="author")classArticle(Base):__tablename__="article"id=Column(Integer,primary_key=True,autoincrement=True)title=Column(String(200),nullable=False)user_id=Column(Integer,ForeignKey("user.id"))# 通过 relationship 直接访问文章作者author=relationship("User",back_populates="articles")

使用关联：

user=session.query(User).first()print(user.articles)# 该用户的所有文章（自动 JOIN 查询）article=session.query(Article).first()print(article.author.username)# 文章对应的作者用户名

事务管理

Session 中的所有操作在commit()之前都只是内存中的修改，提交后才真正写入数据库。rollback()可以回滚未提交的修改，常用于异常处理。

fromsqlalchemy.excimportSQLAlchemyError session=SessionLocal()try:user=User(username="eve",email="eve@example.com",age=22)session.add(user)session.commit()# 提交事务exceptSQLAlchemyErrorase:session.rollback()# 出错回滚，保证数据一致print("操作失败：",e)finally:session.close()# 无论成败都关闭会话

原生 SQL 混合使用

需要复杂查询或绕过 ORM 时，可以用text()直接执行原生 SQL：

fromsqlalchemyimporttext# 查询result=session.execute(text("SELECT * FROM user WHERE age > :age"),{"age":18})forrowinresult:print(row)# 更新session.execute(text("UPDATE user SET age = :age WHERE id = :id"),{"age":23,"id":1})session.commit()

会话管理最佳实践

1. 用完即关：Session 不是线程安全的，不要跨线程复用，用完及时close()。

2. 推荐用上下文管理器：自动关闭，避免忘记close()导致连接泄漏：

   withSessionLocal()assession:users=session.query(User).all()# 离开 with 代码块自动 close()

3. 不要在全局长期持有 Session：每个请求/任务创建新的 Session，处理完即销毁。

常见问题 FAQ

Q：执行create_all()后数据库没有表？
A：常见原因——模型类没有被 import（SQLAlchemy 不知道有哪些表）；或连接字符串的库名写错。确保先import模型模块再调用create_all()。

Q：echo=True会一直打印 SQL，生产环境要关吗？
A：是的。调试时开启很方便，生产环境建议关闭（或换成日志框架控制），避免日志量过大。

Q：修改了模型字段，为什么表结构没变？
A：create_all()只建表、不修改已存在的表。改表结构请用 Alembic 迁移工具，不要指望它自动同步。

Q：query()和select()该用哪个？
A：本教程用经典的query()API（2.0 仍支持）。SQLAlchemy 2.0 推荐使用select()+session.execute()的新风格，二者可共存，团队统一即可。

Q：循环 import 报错怎么办？
A：把engine/Base/SessionLocal抽到独立的sqlalchemy_config.py，模型类只import Base，业务代码再同时 import 配置和模型，避免相互引用。

小结：SQLAlchemy 让"用 Python 对象操作数据库"成为可能。掌握 Engine / Session / Base / Model / Column 五件套，再吃透 CRUD 与关联，就能应对绝大多数业务场景。