文章目录一、核心概念二、安装三、快速开始SQLAlchemy 2.0 风格1. 创建引擎与基类2. 定义模型3. 创建表四、最常用的 CRUD 操作1. 增加Create2. 查询Read3. 更新Update4. 删除Delete五、关系查询六、FastAPI 中的典型用法七、事务与异常处理八、Alembic 数据库迁移强烈建议九、最佳实践十、Core 与 ORM 如何选择SQLAlchemy 是 Python 中最强大、最流行的 SQL 工具包同时支持ORM对象关系映射和Core核心表达式两种风格。现代项目推荐使用SQLAlchemy 2.0风格API 更清晰、类型提示更好。一、核心概念概念说明Engine数据库连接池的入口负责管理连接SessionORM 的工作单元负责对象的增删改查Model普通 Python 类对应数据库表Core更接近 SQL 的表达式语言性能更高、更灵活ORM用对象操作数据库开发效率高大多数 Web 项目尤其 FastAPI使用 ORM 风格即可。二、安装pipinstallsqlalchemy# 根据数据库选择驱动pipinstallpsycopg2-binary# PostgreSQLpipinstallpymysql# MySQLpipinstallaiosqlite# SQLite 异步可选三、快速开始SQLAlchemy 2.0 风格1. 创建引擎与基类fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker,Mapped,mapped_column# 数据库连接地址DATABASE_URLsqlite:///./test.db# PostgreSQL 示例postgresql://user:passwordlocalhost/dbnameenginecreate_engine(DATABASE_URL,echoTrue,# 打印实际执行的 SQL开发时建议开启)classBase(DeclarativeBase):pass# 创建 Session 工厂SessionLocalsessionmaker(bindengine,autoflushFalse,autocommitFalse)2. 定义模型fromdatetimeimportdatetimefromsqlalchemyimportString,Integer,DateTime,ForeignKey,Textfromsqlalchemy.ormimportMapped,mapped_column,relationshipclassUser(Base):__tablename__usersid:Mapped[int]mapped_column(primary_keyTrue)name:Mapped[str]mapped_column(String(50),indexTrue)email:Mapped[str]mapped_column(String(100),uniqueTrue,indexTrue)age:Mapped[int|None]mapped_column(Integer,nullableTrue)created_at:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now)# 一对多关系posts:Mapped[list[Post]]relationship(back_populatesauthor)classPost(Base):__tablename__postsid:Mapped[int]mapped_column(primary_keyTrue)title:Mapped[str]mapped_column(String(200))content:Mapped[str]mapped_column(Text)user_id:Mapped[int]mapped_column(ForeignKey(users.id))author:Mapped[User]relationship(back_populatesposts)3. 创建表Base.metadata.create_all(bindengine)四、最常用的 CRUD 操作fromsqlalchemyimportselect# 获取 Sessiondefget_db():dbSessionLocal()try:yielddbfinally:db.close()1. 增加CreatedbSessionLocal()userUser(nameAlice,emailaliceexample.com,age25)db.add(user)db.commit()# 提交事务db.refresh(user)# 刷新对象获取数据库生成的 id 等print(user.id)# 批量添加db.add_all([User(nameBob,emailbobexample.com),User(nameCharlie,emailcharlieexample.com),])db.commit()2. 查询Read# 查询所有usersdb.scalars(select(User)).all()# 按主键查询userdb.get(User,1)# 条件查询stmtselect(User).where(User.age20).order_by(User.id.desc())usersdb.scalars(stmt).all()# 查询第一条userdb.scalars(select(User).where(User.emailaliceexample.com)).first()# 只获取部分字段stmtselect(User.name,User.email)resultsdb.execute(stmt).all()3. 更新Updateuserdb.get(User,1)ifuser:user.age26user.nameAlice Updateddb.commit()或者使用update语句适合批量fromsqlalchemyimportupdate stmtupdate(User).where(User.id1).values(age26)db.execute(stmt)db.commit()4. 删除Deleteuserdb.get(User,1)ifuser:db.delete(user)db.commit()五、关系查询# 创建用户和文章userUser(nameAlice,emailaliceexample.com)post1Post(title第一篇文章,content内容...,authoruser)post2Post(title第二篇文章,content内容...,authoruser)db.add(user)# 级联添加 posts需要配置 cascadedb.commit()# 查询用户时加载文章userdb.scalars(select(User).where(User.id1)).first()print(user.posts)# 自动加载相关文章默认 lazy常用关系加载策略lazyselect默认lazyjoined连表一次加载lazyselectin推荐适合一对多六、FastAPI 中的典型用法fromfastapiimportDepends,FastAPI,HTTPExceptionfromsqlalchemy.ormimportSession appFastAPI()defget_db():dbSessionLocal()try:yielddbfinally:db.close()app.post(/users/)defcreate_user(name:str,email:str,db:SessionDepends(get_db)):userUser(namename,emailemail)db.add(user)db.commit()db.refresh(user)returnuserapp.get(/users/{user_id})defread_user(user_id:int,db:SessionDepends(get_db)):userdb.get(User,user_id)ifnotuser:raiseHTTPException(status_code404,detail用户不存在)returnuser七、事务与异常处理dbSessionLocal()try:userUser(nameAlice,emailaliceexample.com)db.add(user)db.commit()exceptException:db.rollback()# 出错回滚raisefinally:db.close()在 FastAPI 的依赖中通常把commit放在业务逻辑成功后异常时自动回滚。八、Alembic 数据库迁移强烈建议直接create_all只适合开发阶段。生产环境应使用Alembic做版本化迁移。pipinstallalembic alembic init alembic常用命令alembic revision--autogenerate-mcreate users tablealembic upgradeheadalembic downgrade-1九、最佳实践使用 SQLAlchemy 2.0 风格Mapped、mapped_column、select()Session 生命周期要短不要长期持有生产环境关闭echoTrue合理使用索引indexTrue、uniqueTrue关系加载注意 N1 问题必要时用selectinload或joinedload密码等敏感字段不要明文存储大型项目推荐把 Model、Session、CRUD 分层优先使用 Alembic 管理表结构变更十、Core 与 ORM 如何选择ORM开发速度快适合大多数业务系统Core需要极致性能、复杂 SQL、批量操作时更合适两者可以在同一项目中混合使用。 感谢阅读想了解更多 我的博客网站 | 记录思考分享干货 我的个人主页 | 关于我、开源项目 Python SQLAlchemy 与数据库操作讲解
文章目录一、核心概念二、安装三、快速开始SQLAlchemy 2.0 风格1. 创建引擎与基类2. 定义模型3. 创建表四、最常用的 CRUD 操作1. 增加Create2. 查询Read3. 更新Update4. 删除Delete五、关系查询六、FastAPI 中的典型用法七、事务与异常处理八、Alembic 数据库迁移强烈建议九、最佳实践十、Core 与 ORM 如何选择SQLAlchemy 是 Python 中最强大、最流行的 SQL 工具包同时支持ORM对象关系映射和Core核心表达式两种风格。现代项目推荐使用SQLAlchemy 2.0风格API 更清晰、类型提示更好。一、核心概念概念说明Engine数据库连接池的入口负责管理连接SessionORM 的工作单元负责对象的增删改查Model普通 Python 类对应数据库表Core更接近 SQL 的表达式语言性能更高、更灵活ORM用对象操作数据库开发效率高大多数 Web 项目尤其 FastAPI使用 ORM 风格即可。二、安装pipinstallsqlalchemy# 根据数据库选择驱动pipinstallpsycopg2-binary# PostgreSQLpipinstallpymysql# MySQLpipinstallaiosqlite# SQLite 异步可选三、快速开始SQLAlchemy 2.0 风格1. 创建引擎与基类fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker,Mapped,mapped_column# 数据库连接地址DATABASE_URLsqlite:///./test.db# PostgreSQL 示例postgresql://user:passwordlocalhost/dbnameenginecreate_engine(DATABASE_URL,echoTrue,# 打印实际执行的 SQL开发时建议开启)classBase(DeclarativeBase):pass# 创建 Session 工厂SessionLocalsessionmaker(bindengine,autoflushFalse,autocommitFalse)2. 定义模型fromdatetimeimportdatetimefromsqlalchemyimportString,Integer,DateTime,ForeignKey,Textfromsqlalchemy.ormimportMapped,mapped_column,relationshipclassUser(Base):__tablename__usersid:Mapped[int]mapped_column(primary_keyTrue)name:Mapped[str]mapped_column(String(50),indexTrue)email:Mapped[str]mapped_column(String(100),uniqueTrue,indexTrue)age:Mapped[int|None]mapped_column(Integer,nullableTrue)created_at:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now)# 一对多关系posts:Mapped[list[Post]]relationship(back_populatesauthor)classPost(Base):__tablename__postsid:Mapped[int]mapped_column(primary_keyTrue)title:Mapped[str]mapped_column(String(200))content:Mapped[str]mapped_column(Text)user_id:Mapped[int]mapped_column(ForeignKey(users.id))author:Mapped[User]relationship(back_populatesposts)3. 创建表Base.metadata.create_all(bindengine)四、最常用的 CRUD 操作fromsqlalchemyimportselect# 获取 Sessiondefget_db():dbSessionLocal()try:yielddbfinally:db.close()1. 增加CreatedbSessionLocal()userUser(nameAlice,emailaliceexample.com,age25)db.add(user)db.commit()# 提交事务db.refresh(user)# 刷新对象获取数据库生成的 id 等print(user.id)# 批量添加db.add_all([User(nameBob,emailbobexample.com),User(nameCharlie,emailcharlieexample.com),])db.commit()2. 查询Read# 查询所有usersdb.scalars(select(User)).all()# 按主键查询userdb.get(User,1)# 条件查询stmtselect(User).where(User.age20).order_by(User.id.desc())usersdb.scalars(stmt).all()# 查询第一条userdb.scalars(select(User).where(User.emailaliceexample.com)).first()# 只获取部分字段stmtselect(User.name,User.email)resultsdb.execute(stmt).all()3. 更新Updateuserdb.get(User,1)ifuser:user.age26user.nameAlice Updateddb.commit()或者使用update语句适合批量fromsqlalchemyimportupdate stmtupdate(User).where(User.id1).values(age26)db.execute(stmt)db.commit()4. 删除Deleteuserdb.get(User,1)ifuser:db.delete(user)db.commit()五、关系查询# 创建用户和文章userUser(nameAlice,emailaliceexample.com)post1Post(title第一篇文章,content内容...,authoruser)post2Post(title第二篇文章,content内容...,authoruser)db.add(user)# 级联添加 posts需要配置 cascadedb.commit()# 查询用户时加载文章userdb.scalars(select(User).where(User.id1)).first()print(user.posts)# 自动加载相关文章默认 lazy常用关系加载策略lazyselect默认lazyjoined连表一次加载lazyselectin推荐适合一对多六、FastAPI 中的典型用法fromfastapiimportDepends,FastAPI,HTTPExceptionfromsqlalchemy.ormimportSession appFastAPI()defget_db():dbSessionLocal()try:yielddbfinally:db.close()app.post(/users/)defcreate_user(name:str,email:str,db:SessionDepends(get_db)):userUser(namename,emailemail)db.add(user)db.commit()db.refresh(user)returnuserapp.get(/users/{user_id})defread_user(user_id:int,db:SessionDepends(get_db)):userdb.get(User,user_id)ifnotuser:raiseHTTPException(status_code404,detail用户不存在)returnuser七、事务与异常处理dbSessionLocal()try:userUser(nameAlice,emailaliceexample.com)db.add(user)db.commit()exceptException:db.rollback()# 出错回滚raisefinally:db.close()在 FastAPI 的依赖中通常把commit放在业务逻辑成功后异常时自动回滚。八、Alembic 数据库迁移强烈建议直接create_all只适合开发阶段。生产环境应使用Alembic做版本化迁移。pipinstallalembic alembic init alembic常用命令alembic revision--autogenerate-mcreate users tablealembic upgradeheadalembic downgrade-1九、最佳实践使用 SQLAlchemy 2.0 风格Mapped、mapped_column、select()Session 生命周期要短不要长期持有生产环境关闭echoTrue合理使用索引indexTrue、uniqueTrue关系加载注意 N1 问题必要时用selectinload或joinedload密码等敏感字段不要明文存储大型项目推荐把 Model、Session、CRUD 分层优先使用 Alembic 管理表结构变更十、Core 与 ORM 如何选择ORM开发速度快适合大多数业务系统Core需要极致性能、复杂 SQL、批量操作时更合适两者可以在同一项目中混合使用。 感谢阅读想了解更多 我的博客网站 | 记录思考分享干货 我的个人主页 | 关于我、开源项目