SQLAlchemy是Python中最流行的ORM对象关系映射框架之一它提供了高效且灵活的数据库操作方式。本文将介绍如何使用SQLAlchemy ORM进行数据库操作。目录安装SQLAlchemy核心概念连接数据库定义数据模型创建数据库表基本CRUD操作查询数据关系操作事务管理最佳实践安装bashpip install sqlalchemy如果需要连接特定数据库还需安装相应的驱动程序bash# PostgreSQLpip install psycopg2-binary# MySQLpip install mysql-connector-python# SQLite (Python标准库已包含无需额外安装)核心概念Engine数据库连接的引擎负责与数据库通信Session数据库会话管理所有持久化操作Model数据模型类对应数据库中的表Query查询对象用于构建和执行数据库查询连接数据库pythonfrom sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmaker# 创建数据库连接引擎# SQLite示例engine create_engine(sqlite:///example.db, echoTrue)# PostgreSQL示例# engine create_engine(postgresql://username:passwordlocalhost:5432/mydatabase)# MySQL示例# engine create_engine(mysqlmysqlconnector://username:passwordlocalhost:3306/mydatabase)# 创建会话工厂SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine)# 创建会话实例session SessionLocal()定义数据模型pythonfrom sqlalchemy import Column, Integer, String, ForeignKeyfrom sqlalchemy.orm import relationship, declarative_base# 创建基类Base declarative_base()class User(Base):__tablename__ usersid Column(Integer, primary_keyTrue, indexTrue)name Column(String(50), nullableFalse)email Column(String(100), uniqueTrue, indexTrue)# 定义一对多关系posts relationship(Post, back_populatesauthor)class Post(Base):__tablename__ postsid Column(Integer, primary_keyTrue, indexTrue)title Column(String(100), nullableFalse)content Column(String(500))author_id Column(Integer, ForeignKey(users.id))# 定义多对一关系author relationship(User, back_populatesposts)# 定义多对多关系通过关联表tags relationship(Tag, secondarypost_tags, back_populatesposts)class Tag(Base):__tablename__ tagsid Column(Integer, primary_keyTrue, indexTrue)name Column(String(30), uniqueTrue, nullableFalse)posts relationship(Post, secondarypost_tags, back_populatestags)# 关联表用于多对多关系class PostTag(Base):__tablename__ post_tagspost_id Column(Integer, ForeignKey(posts.id), primary_keyTrue)tag_id Column(Integer, ForeignKey(tags.id), primary_keyTrue)创建数据库表python# 创建所有表Base.metadata.create_all(bindengine)# 删除所有表# Base.metadata.drop_all(bindengine)基本CRUD操作创建数据python# 创建新用户new_user User(name张三, emailzhangsanexample.com)session.add(new_user)session.commit()# 批量创建session.add_all([User(name李四, emaillisiexample.com),User(name王五, emailwangwuexample.com)])session.commit()读取数据python# 获取所有用户users session.query(User).all()# 获取第一个用户first_user session.query(User).first()# 根据ID获取用户user session.query(User).get(1)更新数据python# 查询并更新user session.query(User).get(1)user.name 张三四session.commit()# 批量更新session.query(User).filter(User.name.like(张%)).update({name: 张氏}, synchronize_sessionFalse)session.commit()删除数据python# 查询并删除user session.query(User).get(1)session.delete(user)session.commit()# 批量删除session.query(User).filter(User.name 李四).delete(synchronize_sessionFalse)session.commit()查询数据基本查询python# 获取所有记录users session.query(User).all()# 获取特定字段names session.query(User.name).all()# 排序users session.query(User).order_by(User.name.desc()).all()# 限制结果数量users session.query(User).limit(10).all()# 偏移量users session.query(User).offset(5).limit(10).all()过滤查询pythonfrom sqlalchemy import or_# 等值过滤user session.query(User).filter(User.name 张三).first()# 模糊查询users session.query(User).filter(User.name.like(张%)).all()# IN查询users session.query(User).filter(User.name.in_([张三, 李四])).all()# 多条件查询users session.query(User).filter(User.name 张三,User.email.like(%example.com)).all()# 或条件users session.query(User).filter(or_(User.name 张三, User.name 李四)).all()# 不等于users session.query(User).filter(User.name ! 张三).all()聚合查询pythonfrom sqlalchemy import func# 计数count session.query(User).count()# 分组计数user_post_count session.query(User.name,func.count(Post.id)).join(Post).group_by(User.name).all()# 求和、平均值等avg_id session.query(func.avg(User.id)).scalar()连接查询python# 内连接results session.query(User, Post).join(Post).filter(Post.title.like(%Python%)).all()# 左外连接results session.query(User, Post).outerjoin(Post).all()# 指定连接条件results session.query(User, Post).join(Post, User.id Post.author_id).all()关系操作python# 创建带关系的对象user User(name赵六, emailzhaoliuexample.com)post Post(title我的第一篇博客, contentHello World!, authoruser)session.add(post)session.commit()# 通过关系访问print(f文章 {post.title} 的作者是 {post.author.name})print(f用户 {user.name} 的所有文章:)for p in user.posts:print(f - {p.title})# 多对多关系操作python_tag Tag(namePython)sqlalchemy_tag Tag(nameSQLAlchemy)post.tags.append(python_tag)post.tags.append(sqlalchemy_tag)session.commit()print(f文章 {post.title} 的标签:)for tag in post.tags:print(f - {tag.name})事务管理python# 自动提交事务try:user User(name测试用户, emailtestexample.com)session.add(user)session.commit()except Exception as e:session.rollback()print(f发生错误: {e})# 使用事务上下文管理器from sqlalchemy.orm import Sessiondef create_user(session: Session, name: str, email: str):try:user User(namename, emailemail)session.add(user)session.commit()return userexcept:session.rollback()raise# 嵌套事务with session.begin_nested():user User(name事务用户, emailtransactionexample.com)session.add(user)# 保存点savepoint session.begin_nested()try:user User(name保存点用户, emailsavepointexample.com)session.add(user)savepoint.commit()except:savepoint.rollback()最佳实践会话管理为每个请求创建新会话请求结束后关闭异常处理始终处理异常并适当回滚事务延迟加载注意N1查询问题使用 eager loading 优化连接池合理配置连接池大小和超时设置数据验证在模型层或应用层验证数据完整性python# 使用上下文管理器管理会话from contextlib import contextmanagercontextmanagerdef get_db():db SessionLocal()try:yield dbdb.commit()except Exception:db.rollback()raisefinally:db.close()# 使用示例with get_db() as db:user User(name上下文用户, emailcontextexample.com)db.add(user)总结SQLAlchemy ORM提供了强大而灵活的数据库操作方式通过本文的介绍您应该能够安装和配置SQLAlchemy定义数据模型和关系执行基本的CRUD操作构建复杂查询管理数据库事务遵循最佳实践SQLAlchemy还有更多高级特性如混合属性、事件监听、自定义查询等值得进一步探索学习。https://github.com/jampga/hmumv/issues/957https://github.com/jampga/hmumv/issues/956https://github.com/jampga/hmumv/issues/955https://github.com/jampga/hmumv/issues/954https://github.com/jampga/hmumv/issues/953https://github.com/jampga/hmumv/issues/952https://github.com/jampga/hmumv/issues/951https://github.com/jampga/hmumv/issues/950https://github.com/jampga/hmumv/issues/949https://github.com/jampga/hmumv/issues/948https://github.com/jampga/hmumv/issues/947https://github.com/jampga/hmumv/issues/946https://github.com/jampga/hmumv/issues/945https://github.com/jampga/hmumv/issues/944https://github.com/jampga/hmumv/issues/943https://github.com/jampga/hmumv/issues/942https://github.com/jampga/hmumv/issues/941https://github.com/jampga/hmumv/issues/940https://github.com/jampga/hmumv/issues/939https://github.com/jampga/hmumv/issues/938https://github.com/jampga/hmumv/issues/937https://github.com/jampga/hmumv/issues/936https://github.com/jampga/hmumv/issues/935https://github.com/jampga/hmumv/issues/934https://github.com/jampga/hmumv/issues/933https://github.com/jampga/hmumv/issues/932https://github.com/jampga/hmumv/issues/931https://github.com/jampga/hmumv/issues/930https://github.com/jampga/hmumv/issues/929https://github.com/jampga/hmumv/issues/928https://github.com/jampga/hmumv/issues/927https://github.com/jampga/hmumv/issues/926https://github.com/jampga/hmumv/issues/925https://github.com/jampga/hmumv/issues/924https://github.com/jampga/hmumv/issues/923https://github.com/jampga/hmumv/issues/922https://github.com/jampga/hmumv/issues/921https://github.com/jampga/hmumv/issues/920https://github.com/jampga/hmumv/issues/919https://github.com/jampga/hmumv/issues/918https://github.com/jampga/hmumv/issues/914https://github.com/jampga/hmumv/issues/917https://github.com/jampga/hmumv/issues/916https://github.com/jampga/hmumv/issues/913https://github.com/jampga/hmumv/issues/915https://github.com/jampga/hmumv/issues/912https://github.com/jampga/hmumv/issues/911https://github.com/jampga/hmumv/issues/910https://github.com/jampga/hmumv/issues/908https://github.com/jampga/hmumv/issues/909https://github.com/jampga/hmumv/issues/907https://github.com/jampga/hmumv/issues/906https://github.com/jampga/hmumv/issues/904https://github.com/jampga/hmumv/issues/905https://github.com/jampga/hmumv/issues/903https://github.com/jampga/hmumv/issues/902https://github.com/jampga/hmumv/issues/901https://github.com/jampga/hmumv/issues/900https://github.com/jampga/hmumv/issues/899https://github.com/jampga/hmumv/issues/898https://github.com/jampga/hmumv/issues/897https://github.com/jampga/hmumv/issues/895https://github.com/jampga/hmumv/issues/896https://github.com/jampga/hmumv/issues/894https://github.com/jampga/hmumv/issues/893https://github.com/jampga/hmumv/issues/892https://github.com/jampga/hmumv/issues/890https://github.com/jampga/hmumv/issues/889https://github.com/jampga/hmumv/issues/891https://github.com/jampga/hmumv/issues/888https://github.com/jampga/hmumv/issues/887https://github.com/jampga/hmumv/issues/886https://github.com/jampga/hmumv/issues/885https://github.com/jampga/hmumv/issues/884https://github.com/jampga/hmumv/issues/883https://github.com/jampga/hmumv/issues/882https://github.com/jampga/hmumv/issues/881https://github.com/jampga/hmumv/issues/880https://github.com/jampga/hmumv/issues/879