Python 课程18-SQLAlchemy

前言

SQLAlchemy 是一个功能强大的 Python SQL 工具包和对象关系映射（ORM）库，它使得开发者能够通过 Python 代码与数据库进行交互，而不必编写 SQL 查询。SQLAlchemy 提供了对多种数据库的支持，包括 MySQL、PostgreSQL、SQLite 等，适用于从简单的小项目到复杂的大型系统。

本教程将带你从 SQLAlchemy 的基础操作（如连接数据库、创建模型、执行查询等）到高级功能（如事务管理、关系映射等），并提供详细的代码示例。

目录

1. SQLAlchemy 基础

   • 安装 SQLAlchemy
   • 创建数据库连接
   • 使用 SQLAlchemy Core 执行原生 SQL 查询

2. ORM 基础

   • 定义模型（Classes as Tables）
   • 创建表结构
   • 插入、查询、更新、删除数据

3. 关系映射

   • 一对多关系
   • 多对多关系
   • 级联操作

4. 事务与连接池

   • 事务管理
   • 使用连接池提高性能

5. 高级功能

   • 查询构造器与过滤器
   • 自定义查询与聚合操作

1. SQLAlchemy 基础

安装 SQLAlchemy

通过 pip 安装 SQLAlchemy：

pip install sqlalchemy

对于 MySQL 或 PostgreSQL 这样的数据库，你还需要安装相应的驱动程序：

pip install pymysql  # 对于 MySQL
pip install psycopg2  # 对于 PostgreSQL

创建数据库连接

SQLAlchemy 的基础在于创建与数据库的连接，你可以通过 create_engine() 函数来创建引擎对象，它代表了数据库连接的核心。

• 连接 SQLite 数据库（SQLite 是一个轻量级数据库，适用于小型项目）：

from sqlalchemy import create_engine

# 创建 SQLite 引擎
engine = create_engine('sqlite:///example.db', echo=True)

•  连接 MySQL 数据库：

engine = create_engine('mysql+pymysql://username:password@localhost/mydatabase')

•  连接 PostgreSQL 数据库：

engine = create_engine('postgresql+psycopg2://username:password@localhost/mydatabase')

echo=True 会打印生成的 SQL 语句，帮助调试。

使用 SQLAlchemy Core 执行原生 SQL 查询

除了 ORM，SQLAlchemy 还提供了 Core API，用于直接执行 SQL 查询。

• 创建表：

from sqlalchemy import MetaData, Table, Column, Integer, String

metadata = MetaData()

# 定义表结构
users_table = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    Column('age', Integer)
)

# 创建表
metadata.create_all(engine)

• 插入数据：

from sqlalchemy import insert

# 插入数据
stmt = insert(users_table).values(name='Alice', age=25)
with engine.connect() as conn:
    conn.execute(stmt)

•  查询数据：

from sqlalchemy import select

# 查询数据
stmt = select(users_table)
with engine.connect() as conn:
    result = conn.execute(stmt)
    for row in result:
        print(row)

• 更新与删除数据：

from sqlalchemy import update, delete

# 更新数据
stmt = update(users_table).where(users_table.c.name == 'Alice').values(age=30)
with engine.connect() as conn:
    conn.execute(stmt)

# 删除数据
stmt = delete(users_table).where(users_table.c.name == 'Alice')
with engine.connect() as conn:
    conn.execute(stmt)

2. ORM 基础

定义模型（Classes as Tables）

在 SQLAlchemy ORM 中，表结构通过 Python 类表示。每个类代表数据库中的一张表，类的属性代表表中的列。

• 定义模型类：

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'  # 表名
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

    def __repr__(self):
        return f"<User(name={self.name}, age={self.age})>"

创建表结构

使用 Base.metadata.create_all() 创建模型类对应的表结构。

Base.metadata.create_all(engine)

插入、查询、更新、删除数据

• 创建数据库会话：

为了与数据库交互，SQLAlchemy 使用会话（Session）对象。它是数据库连接的一个高层次接口。

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

•  插入数据：

# 插入数据
new_user = User(name='Bob', age=22)
session.add(new_user)
session.commit()

•  查询数据：

# 查询所有用户
users = session.query(User).all()
for user in users:
    print(user)

# 查询特定条件的数据
user = session.query(User).filter_by(name='Bob').first()
print(user)

•  更新数据：

# 更新 Bob 的年龄
user = session.query(User).filter_by(name='Bob').first()
user.age = 23
session.commit()

• 删除数据：

# 删除用户
user = session.query(User).filter_by(name='Bob').first()
session.delete(user)
session.commit()

3. 关系映射

一对多关系

在数据库中，一对多关系是非常常见的。可以使用 SQLAlchemy 定义这样的关系，例如，一个用户可以有多个地址。

• 定义一对多关系：

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))

    user = relationship('User', back_populates='addresses')

User.addresses = relationship('Address', order_by=Address.id, back_populates='user')

在这个模型中，User 与 Address 之间建立了一对多的关系。

• 插入与查询关系数据：

new_user = User(name='Charlie', age=30)
new_address = Address(email='charlie@example.com', user=new_user)

session.add(new_user)
session.add(new_address)
session.commit()

# 查询用户及其地址
user = session.query(User).filter_by(name='Charlie').first()
print(user.addresses)  # 输出用户的地址列表

多对多关系

在多对多关系中，两个表之间通过一个中间表来关联。

• 定义多对多关系：

from sqlalchemy import Table

association_table = Table('association', Base.metadata,
    Column('user_id', Integer, ForeignKey('users.id')),
    Column('group_id', Integer, ForeignKey('groups.id'))
)

class Group(Base):
    __tablename__ = 'groups'
    id = Column(Integer, primary_key=True)
    name = Column(String)

User.groups = relationship('Group', secondary=association_table, back_populates='users')
Group.users = relationship('User', secondary=association_table, back_populates='groups')

•  插入与查询多对多关系数据：

# 创建用户与群组
new_group = Group(name='Admins')
new_user = User(name='Alice', age=25)
new_user.groups.append(new_group)

session.add(new_user)
session.commit()

# 查询用户的群组
user = session.query(User).filter_by(name='Alice').first()
print(user.groups)  # 输出用户所在的群组

级联操作

级联操作可以在删除或更新主表数据时自动影响相关的表。可以通过 cascade 参数来控制级联行为。

• 定义级联删除：

User.addresses = relationship('Address', back_populates='user', cascade='all, delete-orphan')

此设置意味着如果删除一个 User，它的 Address 记录也会被删除。

4. 事务与连接池

事务管理

在数据库操作中，事务管理非常重要，尤其是在处理批量插入、更新和删除时。

• 手动管理事务：

session = Session()

try:
    new_user = User(name='David', age=35)
    session.add(new_user)
    session.commit()
except:
    session.rollback()  # 如果出现错误，回滚事务
    raise
finally:
    session.close()

使用连接池提高性能

SQLAlchemy 提供了对数据库连接池的支持，以提高数据库访问的性能。你可以通过在创建引擎时指定连接池配置来管理连接。

• 设置连接池：

engine = create_engine(
    'mysql+pymysql://username:password@localhost/mydatabase',
    pool_size=5,  # 连接池的大小
    max_overflow=10,  # 当连接池用尽时，最多允许额外创建的连接数
    pool_timeout=30,  # 等待连接池的超时时间（秒）
    pool_recycle=3600  # 每隔一小时回收一次连接，以避免长时间的空闲连接
)

这种配置可以防止频繁建立和关闭数据库连接，尤其在需要高效访问数据库的场景中极为重要。

5. 高级功能

查询构造器与过滤器

SQLAlchemy ORM 提供了丰富的查询构造功能，使得我们能够以面向对象的方式生成复杂的查询。以下是一些常用的查询构造方式：

• 查询所有记录：

users = session.query(User).all()
for user in users:
    print(user)

• 过滤查询：

# 按名字过滤
users = session.query(User).filter_by(name='Alice').all()

# 使用条件表达式
users = session.query(User).filter(User.age > 30).all()

•  排序与限制：

# 按年龄排序
users = session.query(User).order_by(User.age).all()

# 只返回前 5 个用户
users = session.query(User).limit(5).all()

•  联接查询（查询多个表）：

# 查询用户和他们的地址
results = session.query(User, Address).join(Address).all()
for user, address in results:
    print(f'{user.name} lives at {address.email}')

• 计数、求和与聚合操作：

from sqlalchemy import func

# 计算用户数量
user_count = session.query(func.count(User.id)).scalar()

# 计算用户的平均年龄
average_age = session.query(func.avg(User.age)).scalar()

自定义查询与聚合操作

SQLAlchemy 的 func 模块使得我们能够使用数据库中的聚合函数，如 COUNT、SUM、MAX 等。

• 聚合查询：

# 查询用户的最大年龄
max_age = session.query(func.max(User.age)).scalar()

# 计算特定条件下的总人数
count = session.query(func.count(User.id)).filter(User.age > 30).scalar()

原生 SQL 查询

如果需要执行复杂的原生 SQL 查询，SQLAlchemy 也提供了直接执行原生 SQL 的能力。

• 执行原生 SQL：

result = session.execute('SELECT * FROM users WHERE age > :age', {'age': 30})
for row in result:
    print(row)

通过这种方式，你可以自由使用数据库特有的 SQL 语句。

结论

        通过本教程，你已经详细了解了 SQLAlchemy 的基本与高级功能，从建立数据库连接、创建表结构、到复杂的查询与事务管理等。SQLAlchemy 提供了两种主要的使用模式：

• SQLAlchemy Core：用于执行原生 SQL 操作，适用于需要精准控制数据库查询的场景。
• SQLAlchemy ORM：为开发者提供了更加 Pythonic 的方式来管理数据库模型，隐藏了 SQL 复杂性，更适合业务逻辑开发。