SQLAlchemy是Python中最流行的ORM对象关系映射框架之一它提供了高效且灵活的数据库操作方式。本文将介绍如何使用SQLAlchemy ORM进行数据库操作。目录安装SQLAlchemy核心概念连接数据库定义数据模型创建数据库表基本CRUD操作查询数据关系操作事务管理最佳实践安装bashpip install sqlalchemy如果需要连接特定数据库还需安装相应的驱动程序bash# PostgreSQL pip install psycopg2-binary # MySQL pip install mysql-connector-python # SQLite (Python标准库已包含无需额外安装)核心概念Engine数据库连接的引擎负责与数据库通信Session数据库会话管理所有持久化操作Model数据模型类对应数据库中的表Query查询对象用于构建和执行数据库查询连接数据库pythonfrom sqlalchemy import create_engine from 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, ForeignKey from sqlalchemy.orm import relationship, declarative_base # 创建基类 Base declarative_base() class User(Base): __tablename__ users id 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__ posts id 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__ tags id Column(Integer, primary_keyTrue, indexTrue) name Column(String(30), uniqueTrue, nullableFalse) posts relationship(Post, secondarypost_tags, back_populatestags) # 关联表用于多对多关系 class PostTag(Base): __tablename__ post_tags post_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 Session def create_user(session: Session, name: str, email: str): try: user User(namename, emailemail) session.add(user) session.commit() return user except: 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 contextmanager contextmanager def get_db(): db SessionLocal() try: yield db db.commit() except Exception: db.rollback() raise finally: db.close() # 使用示例 with get_db() as db: user User(name上下文用户, emailcontextexample.com) db.add(user)总结SQLAlchemy ORM提供了强大而灵活的数据库操作方式通过本文的介绍您应该能够安装和配置SQLAlchemy定义数据模型和关系执行基本的CRUD操作构建复杂查询管理数据库事务遵循最佳实践SQLAlchemy还有更多高级特性如混合属性、事件监听、自定义查询等值得进一步探索学习。