SQLAlchemy 学习与使用什么是 SQLAlchemy英 /ˈælkəmi/ —— “艾尔-克-米”原意是炼金术。SQLAlchemy 是 Python 中最流行的ORMObject-Relational Mapping对象关系映射框架之一它将 Python 对象与数据库表进行映射允许开发者使用 Python 语法操作数据库无需直接编写原生 SQL 语句也支持原生 SQL 混合使用大幅提升数据库操作的简洁性、可读性和可维护性。核心优势跨数据库兼容性支持 MySQL、PostgreSQL、SQLite、Oracle 等主流数据库灵活的查询方式强大的事务支持完善的 ORM 映射机制适用于从小型项目到大型企业级应用的各类场景。SQLAlchemy 核心概念SQLAlchemy 的核心分为两大模块Core核心组件和ORM对象关系映射。Core是底层的数据库交互引擎提供 SQL 表达式构建、连接管理等功能ORM是在 Core 之上的封装实现对象与数据库表的映射是我们学习的重点。Core 核心组件Engine引擎SQLAlchemy 的核心入口负责管理数据库连接通过连接字符串创建决定了数据库类型、地址、账号密码等信息。Session会话用于与数据库进行交互的桥梁所有数据库操作增删改查都通过 Session 完成相当于一个临时的数据库连接会话。Base基类所有 ORM 模型类的父类通过 Base 类创建的子类会自动映射为数据库中的表。Model模型Python 中的类对应数据库中的一张表类的属性对应表的字段类的实例对应表的一行数据。Column字段用于定义模型类的属性即数据库表的字段可指定字段类型、主键、非空、默认值等约束。ORM 对象关系映射ORMObject-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 官方驱动其他数据库驱动安装PostgreSQLpip install psycopg2-binarySQLite无需额外安装Python 自带Oraclepip install cx_Oracle安装完成后在 Python 环境中执行以下代码无报错则说明安装成功importsqlalchemyimportpymysqlprint(sqlalchemy.__version__)# 打印 SQLAlchemy 版本号如 2.0.49print(pymysql.__version__)# 打印 pymysql 版本号如 2.2.82. 数据库准备在开始使用 SQLAlchemy 操作 MySQL 前需提前准备好 MySQL 数据库手动创建后续 SQLAlchemy 仅操作表不创建数据库CREATEDATABASEsqlalchemy_demoCHARACTERSETutf8mb4;3. 创建 Engine连接数据库首先通过连接字符串创建 Engine连接字符串的格式根据数据库类型不同而不同fromsqlalchemyimportcreate_engine# MySQL pymysqlenginecreate_engine(mysqlpymysql://root:123456localhost:3306/sqlalchemy_demo?charsetutf8mb4,echoTrue,# 开启 SQL 打印调试时可清晰看到 SQLAlchemy 生成的原生 SQLpool_pre_pingTrue,# 每次取连接前先探活避免拿到失效连接)# 其他数据库的连接字符串示例# SQLite: create_engine(sqlite:///./demo.db)# PostgreSQL: create_engine(postgresqlpsycopg2://user:passlocalhost:5432/demo)说明root数据库用户名替换为自己的数据库账号123456数据库密码替换为自己的密码sqlalchemy_demo要连接的数据库名需提前在 MySQL 中创建echoTrue开启 SQL 打印调试时可清晰看到 SQLAlchemy 生成的原生 SQL便于排查问题4. 创建 Base 基类和 Model 模型通过 Base 基类创建模型类映射到数据库中的表以用户表 user为例fromsqlalchemy.ormimportdeclarative_basefromsqlalchemyimportColumn,Integer,String,DateTime,Float,Boolean Basedeclarative_base()classUser(Base):__tablename__user# 数据库中的表名idColumn(Integer,primary_keyTrue,autoincrementTrue,comment用户ID)usernameColumn(String(50),uniqueTrue,nullableFalse,comment用户名)emailColumn(String(100),nullableFalse,comment邮箱)ageColumn(Integer,comment年龄)created_atColumn(DateTime,server_defaultnow(),comment创建时间)def__repr__(self):returnfUser(id{self.id}, username{self.username})字段类型说明常用Integer整数类型对应 MySQL 的INTString(n)字符串类型n 为最大长度对应 MySQL 的VARCHAR(n)DateTime时间类型对应 MySQL 的DATETIMEFloat浮点数类型对应 MySQL 的FLOAT/DOUBLEBoolean布尔类型对应 MySQL 的TINYINT(1)字段约束说明常用primary_keyTrue设为主键autoincrementTrue自增仅整数主键可用nullableFalse非空约束uniqueTrue唯一约束default默认值comment字段注释Column常用属性表属性类型说明primary_keybool是否为主键True 是autoincrementbool / str是否自增True/False/“auto”commentstr数据库字段注释nullablebool是否允许为空默认 Trueuniquebool是否添加唯一约束indexbool是否创建普通索引defaultAnyPython 代码层面默认值server_default表达式 / 字符串数据库层面默认值如now()foreign_keyForeignKey / str外键约束onupdateAny更新数据时自动赋值ondeletestr外键删除规则CASCADE/SET NULL 等namestr数据库真实字段名不指定则用属性名5. 创建数据表通过 Base 类的create_all()方法自动根据模型类创建数据库表如果表已存在则不会重复创建Base.metadata.create_all(engine)执行后会在sqlalchemy_demo数据库中创建user表可通过 MySQL 客户端查看表结构验证创建结果。6. 创建 Session实现数据库操作Session 是与数据库交互的核心所有操作都需通过 Session 完成步骤为创建 Session → 执行操作 → 提交事务 → 关闭 Session。fromsqlalchemy.ormimportSession sessionSession(engine)7. 插入数据# 单条插入userUser(usernamealice,emailaliceexample.com,age20)session.add(user)session.commit()print(user.id)# 提交后主键会回填# 批量插入session.add_all([User(usernamebob,emailbobexample.com,age25),User(usernamecarol,emailcarolexample.com,age30),])session.commit()8. 查询数据# 查询全部userssession.query(User).all()# 查询第一条firstsession.query(User).first()# 按主键查询onesession.get(User,1)# 推荐写法# 兼容写法session.query(User).get(1)快速开始练习到这一步你已经跑通了建表 → 插入 → 查询的最小闭环。下面我们把它组织成工程化的结构正式进入 CRUD 实战。ORM 核心操作 CRUD抽取通用配置将配置抽象书写在sqlalchemy_config.py中统一定义与使用# sqlalchemy_config.pyfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportdeclarative_base,sessionmaker DB_URImysqlpymysql://root:123456localhost:3306/sqlalchemy_demo?charsetutf8mb4enginecreate_engine(DB_URI,echoTrue,pool_pre_pingTrue)Basedeclarative_base()SessionLocalsessionmaker(bindengine)# 后续用 SessionLocal() 创建会话模型类# models.pyfromsqlalchemyimportColumn,Integer,String,DateTimefromsqlalchemy_configimportBaseclassUser(Base):__tablename__useridColumn(Integer,primary_keyTrue,autoincrementTrue,comment用户ID)usernameColumn(String(50),uniqueTrue,nullableFalse,comment用户名)emailColumn(String(100),nullableFalse,comment邮箱)ageColumn(Integer,comment年龄)def__repr__(self):returnfUser(id{self.id}, username{self.username})CRUD 操作Read查询数据SQLAlchemy 提供了丰富的查询方法核心是query()函数结合过滤条件、排序、分页等操作适配 MySQL 的查询语法。基础查询sessionSessionLocal()# 查询全部all_userssession.query(User).all()# 查询第一条first_usersession.query(User).first()# 按主键查询usersession.get(User,1)条件查询核心是filter()函数添加查询条件支持 MySQL 所有常用运算符# 相等session.query(User).filter(User.usernamealice).first()# 大于 / 小于session.query(User).filter(User.age18).all()session.query(User).filter(User.age60).all()# 多条件 AND逗号分隔即为 ANDsession.query(User).filter(User.age18,User.age30).all()# OR / AND / NOTfromsqlalchemyimportor_,and_,not_ session.query(User).filter(or_(User.age18,User.age60)).all()session.query(User).filter(and_(User.age18,User.age30)).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()常用聚合函数函数作用func.count(字段)统计数量func.sum(字段)求和func.avg(字段)平均值func.max(字段)最大值func.min(字段)最小值Create新增数据新增数据的步骤创建模型实例 → 将实例添加到会话 → 提交会话commit新增后的数据会直接写入 MySQL 数据库。userUser(usernamedave,emaildaveexample.com,age28)session.add(user)session.commit()print(user.id)# 提交后主键自动回填Delete删除数据删除数据的步骤查询数据 → 删除实例 → 提交会话删除操作会直接删除 MySQL 中的对应数据。usersession.query(User).filter(User.usernamedave).first()ifuser:session.delete(user)session.commit()Update更新数据更新数据的步骤查询数据 → 修改实例属性 → 提交会话更新操作会直接同步到 MySQL 数据库。usersession.query(User).filter(User.usernamealice).first()ifuser:user.age21user.emailalice_newexample.comsession.commit()# 修改属性后提交即可无需 add关联关系一对多实际业务中表和表之间往往存在关联。以一个用户拥有多篇文章为例fromsqlalchemyimportColumn,Integer,String,ForeignKeyfromsqlalchemy.ormimportrelationshipfromsqlalchemy_configimportBaseclassUser(Base):__tablename__useridColumn(Integer,primary_keyTrue,autoincrementTrue)usernameColumn(String(50),uniqueTrue,nullableFalse)# 通过 relationship 直接访问该用户的文章articlesrelationship(Article,back_populatesauthor)classArticle(Base):__tablename__articleidColumn(Integer,primary_keyTrue,autoincrementTrue)titleColumn(String(200),nullableFalse)user_idColumn(Integer,ForeignKey(user.id))# 通过 relationship 直接访问文章作者authorrelationship(User,back_populatesarticles)使用关联usersession.query(User).first()print(user.articles)# 该用户的所有文章自动 JOIN 查询articlesession.query(Article).first()print(article.author.username)# 文章对应的作者用户名事务管理Session 中的所有操作在commit()之前都只是内存中的修改提交后才真正写入数据库。rollback()可以回滚未提交的修改常用于异常处理。fromsqlalchemy.excimportSQLAlchemyError sessionSessionLocal()try:userUser(usernameeve,emaileveexample.com,age22)session.add(user)session.commit()# 提交事务exceptSQLAlchemyErrorase:session.rollback()# 出错回滚保证数据一致print(操作失败,e)finally:session.close()# 无论成败都关闭会话原生 SQL 混合使用需要复杂查询或绕过 ORM 时可以用text()直接执行原生 SQLfromsqlalchemyimporttext# 查询resultsession.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()会话管理最佳实践用完即关Session 不是线程安全的不要跨线程复用用完及时close()。推荐用上下文管理器自动关闭避免忘记close()导致连接泄漏withSessionLocal()assession:userssession.query(User).all()# 离开 with 代码块自动 close()不要在全局长期持有 Session每个请求/任务创建新的 Session处理完即销毁。常见问题 FAQQ执行create_all()后数据库没有表A常见原因——模型类没有被 importSQLAlchemy 不知道有哪些表或连接字符串的库名写错。确保先import模型模块再调用create_all()。QechoTrue会一直打印 SQL生产环境要关吗A是的。调试时开启很方便生产环境建议关闭或换成日志框架控制避免日志量过大。Q修改了模型字段为什么表结构没变Acreate_all()只建表、不修改已存在的表。改表结构请用 Alembic 迁移工具不要指望它自动同步。Qquery()和select()该用哪个A本教程用经典的query()API2.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 与关联就能应对绝大多数业务场景。