Python中SQLAlchemy增删改查与表关联应用详解(最新推荐)

目录

• Python中SQLAlchemy的全面学习：增删改查与表关联应用详解
• 引言
• 1. 环境准备
• 安装 SQLAlchemy
• 导入必要的模块
• 2. 创建数据库引擎和会话工厂
• 3. 定义数据模型（表结构）
• 3.1 定义用户表（User）
• 3.2 定义文章表（Post）
• 4. 创建数据库表
• 5. 执行 CRUD 操作（增删改查）
• 5.1 增加（Create）
• 5.2 查询（Read）
• 5.3 修改（Update）
• 5.4 删除（Delete）
• 6. 表之间关联的应用详解
• 6.1 一对一（One-to-One）
• 6.2 一对多（One-to-Many）
• 6.3 多对多（Many-to-Many）
• 7. 高级技巧与最佳实践
• 7.1 事务管理（Transaction Management）
• 7.2 优雅地关闭会话（使用上下文管理器）
• 7.3 使用query的链式调用进行复杂查询
• 总结

SQLAlchemy 是 Python 中最流行的 ORM（对象关系映射）工具之一，它使得开发者能够以面向对象的方式操作数据库，极大地提高了开发效率和代码可读性。本文将详细介绍如何使用 SQLAlchemy 实现数据库的增删改查（CRUD）操作，以及如何处理表之间的关联关系。

pip install sqlalchemy

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, DateTime, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
import datetime

# 创建数据库引擎，这里以 SQLite 为例（也可以使用 MySQL、PostgreSQL 等）
engine = create_engine('sqlite:///example.db', echo=True)  # echo=True 用于显示 SQL 语句
# 创建基类（所有模型类的基类）
Base = declarative_base()
# 创建会话工厂（用于创建会话对象）
SessionLocal = sessionmaker(bind=engine)

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50), nullable=False, unique=True)
    email = Column(String(100), nullable=False)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    # 定义与文章表的关联关系（反向关系）
    posts = relationship("Post", back_populates="author")
    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}', email='{self.email}')>"

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)
    content = Column(String(1000))
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    user_id = Column(Integer, ForeignKey('users.id'))  # 外键关联到 User 表
    # 定义与用户表的关联关系（正向关系）
    author = relationship("User", back_populates="posts")
    def __repr__(self):
        return f"<Post(id={self.id}, title='{self.title}', user_id={self.user_id})>"

# 通过 Base.metadata.create_all() 创建所有定义的表
Base.metadata.create_all(engine)

# 创建会话实例
session = SessionLocal()
try:
    # 创建新用户
    new_user = User(username='alice', email='alice@example.com')
    session.add(new_user)
    session.commit()  # 提交事务，保存到数据库
    print(f"用户 {new_user.username} 添加成功！")
    # 创建新文章，关联到用户
    new_post = Post(title='我的第一篇博客', content='这是我的第一篇文章内容。', user_id=new_user.id)
    session.add(new_post)
    session.commit()
    print(f"文章 {new_post.title} 添加成功！")
except Exception as e:
    session.rollback()  # 出错时回滚事务
    print(f"添加失败：{e}")
finally:
    session.close()

session = SessionLocal()
try:
    user = session.query(User).filter(User.id == 1).first()
    if user:
        print(user)
    else:
        print("未找到该用户")
finally:
    session.close()

session = SessionLocal()
try:
    users = session.query(User).filter(User.username.like('%a%')).all()
    for user in users:
        print(user)
finally:
    session.close()

session = SessionLocal()
try:
    # 查询所有用户的姓名及其发表的文章标题
    results = session.query(User.username, Post.title).join(Post).all()
    for username, title in results:
        print(f"{username} - {title}")
finally:
    session.close()

session = SessionLocal()
try:
    # 获取用户 alice 及其所有文章
    alice = session.query(User).filter(User.username == 'alice').first()
    if alice:
        for post in alice.posts:
            print(post.title)
finally:
    session.close()

session = SessionLocal()
try:
    # 更新用户邮箱
    user = session.query(User).filter(User.username == 'alice').first()
    if user:
        user.email = 'alice_new@example.com'
        session.commit()
        print(f"用户 {user.username} 的邮箱已更新为 {user.email}")
    else:
        print("用户不存在")
except Exception as e:
    session.rollback()
    print(f"更新失败：{e}")
finally:
    session.close()

session = SessionLocal()
try:
    # 删除文章，先删除文章再删除用户（注意外键约束）
    post = session.query(Post).filter(Post.title == '我的第一篇博客').first()
    if post:
        session.delete(post)
        session.commit()
        print(f"文章 {post.title} 已删除")
    else:
        print("文章不存在")
except Exception as e:
    session.rollback()
    print(f"删除失败：{e}")
finally:
    session.close()

# 例如：用户有一个唯一的个人资料卡片（Profile）
class Profile(Base):
    __tablename__ = 'profiles'
    id = Column(Integer, primary_key=True)
    bio = Column(String(500))
    user_id = Column(Integer, ForeignKey('users.id'), unique=True)  # 一对一，外键唯一
    user = relationship("User", backref="profile")

# 已在 User 和 Post 之间实现：一个用户可以有多个文章（一对多）
# 通过 `relationship` 和 `back_populates` 实现双向关系。

# 例如：用户和标签之间是多对多关系（一个用户可以有多个标签，一个标签可以被多个用户使用）
# 创建中间表（关联表）
user_tag = Table('user_tag', Base.metadata,
                Column('user_id', Integer, ForeignKey('users.id')),
                Column('tag_id', Integer, ForeignKey('tags.id'))
               )
class Tag(Base):
    __tablename__ = 'tags'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    # 与用户的关系（多对多）
    users = relationship("User", secondary=user_tag, back_populates="tags")
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50), nullable=False)
    email = Column(String(100), nullable=False)
    # 与标签的关系（多对多）
    tags = relationship("Tag", secondary=user_tag, back_populates="users")

session = SessionLocal()
try:
    # 创建用户和标签
    user = User(username='bob')
    tag1 = Tag(name='Python')
    tag2 = Tag(name='Web')
    # 将标签添加到用户（自动创建中间表记录）
    user.tags.append(tag1)
    user.tags.append(tag2)
    session.add(user)
    session.commit()
    print(f"用户 {user.username} 已添加标签: {[t.name for t in user.tags]}")
except Exception as e:
    session.rollback()
    print(f"添加标签失败：{e}")
finally:
    session.close()

session = SessionLocal()
try:
    # 多个操作在一个事务中执行，保证一致性
    user = User(username='charlie', email='charlie@example.com')
    post = Post(title='测试文章', content='测试内容', user_id=user.id)
    session.add(user)
    session.add(post)
    session.commit()  # 一次性提交，如果出错则回滚
except Exception as e:
    session.rollback()
    print(f"事务失败：{e}")
finally:
    session.close()

from contextlib import contextmanager
@contextmanager
def get_db_session():
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        raise e
    finally:
        session.close()
# 使用示例：
with get_db_session() as session:
    user = User(username='david', email='david@example.com')
    session.add(user)
    # 无需手动调用 commit()，上下文管理器会自动处理

# 查询用户名包含 'a' 且文章数量大于 1 的用户
results = (session.query(User)
           .join(Post)
           .group_by(User.id)
           .having(text("count(Post.id) > 1"))
           .filter(User.username.like('%a%'))
           .all())
for user in results:
    print(f"{user.username} 有 {len(user.posts)} 篇文章")

SQLAlchemy 提供了强大而灵活的 ORM 功能，使我们能够以简洁、直观的方式进行数据库操作。本文系统地介绍了：

• 如何定义模型类并创建数据库表；
• 如何执行基本的 CRUD 操作；
• 如何处理一对一、一对多、多对多等复杂的表关联关系；
• 一些高级技巧，如事务管理、上下文管理器、复杂查询等。

掌握这些知识后，你将能够在实际项目中高效地使用 SQLAlchemy 操控数据库，提升开发效率和代码质量。

到此这篇关于Python中SQLAlchemy增删改查与表关联应用详解(最新推荐)的文章就介绍到这了,更多相关Python SQLAlchemy增删改查内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

• python中的flask_sqlalchemy的使用及示例详解
• Python使用SQLAlchemy操作单表的完整实战指南
• Python库SQLAlchemy的简介、用法和安装步骤详解
• Python ORM神器之SQLAlchemy基本使用完全指南
• python SQLAlchemy 数据库连接池的实现
• Python如何使用sqlalchemy实现动态sql
• Python使用SQLAlchemy操作Mysql数据库的操作示例
• Python中使用sqlalchemy操作数据库的问题总结
• 3个Python SQLAlchemy数据库操作功能详解