Python ORM 框架 SQLAlchemy

1. 为什么选择 SQLAlchemy？

• 灵活性：支持 ORM 和直接 SQL 查询，适合不同场景。
• 跨数据库支持：兼容多种数据库，无需更改代码。
• 高效开发：减少手动编写 SQL 的工作量。
• 社区支持：拥有活跃的社区和丰富的文档。

2. 安装 SQLAlchemy

在开始之前，需要安装 SQLAlchemy 和数据库驱动。以 SQLite 为例（无需额外驱动），MySQL 或 PostgreSQL 需要安装对应的驱动（如 pymysql 或 psycopg2）。

安装步骤

如果使用 PostgreSQL，安装 psycopg2：

pip install psycopg2

如果使用 MySQL，安装 pymysql：

pip install pymysql

使用 pip 安装 SQLAlchemy：

pip install sqlalchemy

验证安装

在 Python 中运行以下代码，确认安装成功：

import sqlalchemy
print(sqlalchemy.__version__)

3. 核心概念

在学习具体操作之前，了解 SQLAlchemy 的核心组件非常重要：

• Engine：数据库连接的入口，管理数据库连接池。
• Connection：用于执行 SQL 语句的连接对象。
• MetaData：存储数据库的表结构信息。
• Table：表示数据库中的表。
• Session：ORM 中用于管理对象与数据库交互的会话。
• Model：Python 类，映射到数据库表。

4. 连接数据库

SQLAlchemy 使用 Engine 来连接数据库。以下是以 SQLite 为例的连接代码：

from sqlalchemy import create_engine

# 创建 Engine，连接到 SQLite 数据库（文件名为 example.db）
engine = create_engine("sqlite:///example.db", echo=True)  # echo=True 打印 SQL 日志

连接其他数据库

• MySQL: create_engine("mysql+pymysql://user:password@localhost/dbname")
• PostgreSQL: create_engine("postgresql+psycopg2://user:password@localhost/dbname")

5. 定义模型（映射表结构）

在 ORM 中，数据库表通过 Python 类定义。以下是定义一个 User 表的示例：

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base

# 创建基类
Base = declarative_base()

# 定义 User 模型
class User(Base):
    __tablename__ = "users"  # 表名
    id = Column(Integer, primary_key=True)  # 主键
    name = Column(String)  # 字符串列
    email = Column(String, unique=True)  # 唯一约束

• declarative_base()：创建 ORM 基类，用于定义模型。
• __tablename__：指定数据库表名。
• Column：定义表中的列，支持多种数据类型（如 Integer、String）。

6. 创建表

定义模型后，需要将表结构创建到数据库中，由于我们在开发过程中可能需要不断的修改表结构，所以为了安全起见，需要我们手动创建表，修改表结构：

# 创建所有定义的表
create table users(
  ......
)

7. 创建会话（Session）

会话用于管理 ORM 对象与数据库的交互：

from sqlalchemy.orm import sessionmaker

# 创建 Session 类
Session = sessionmaker(bind=engine)

# 实例化一个会话
session = Session()

• 每个会话是独立的，操作完成后需关闭（session.close()）。
• 使用 with 语句可以自动管理会话。

8. 基本 CRUD 操作

以下是使用 ORM 进行创建、读取、更新和删除（CRUD）操作的示例。

创建记录

# 创建一个新用户
new_user = User(name="Alice", email="alice@example.com")
session.add(new_user)  # 添加到会话
session.commit()  # 提交到数据库

查询记录

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

# 查询特定用户
user = session.query(User).filter_by(name="Alice").first()
print(user.email)

更新记录

# 更新用户邮箱
user = session.query(User).filter_by(name="Alice").first()
user.email = "alice_new@example.com"
session.commit()

删除记录

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

关系映射

SQLAlchemy 支持表之间的关系（如一对多、多对多）。以下是一个简单的 Post 表与 User 表的一对多关系示例：

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey("users.id"))  # 外键
    user = relationship("User", back_populates="posts")  # 关系

# 在 User 类中添加反向关系
User.posts = relationship("Post", back_populates="user")

查询进阶

• 过滤：session.query(User).filter(User.name.like("%Ali%")).all()
• 排序：session.query(User).order_by(User.name).all()
• 分页：session.query(User).limit(10).offset(20).all()

事务管理

SQLAlchemy 自动管理事务，但可以手动控制：

try:
    session.add(new_user)
    session.commit()
except:
    session.rollback()  # 回滚事务
    raise
finally:
    session.close()  # 关闭会话

9. 最佳实践

• 日志调试：设置 echo=True 查看生成的 SQL，生产环境关闭。
• 连接池管理：SQLAlchemy 默认使用连接池，适合高并发场景。

避免 N+1 问题：使用 joinedload 或 subqueryload 预加载关系。

from sqlalchemy.orm import joinedload
users = session.query(User).options(joinedload(User.posts)).all()

使用上下文管理器：用 with 语句管理会话，自动关闭。

with Session() as session:
    user = User(name="Bob", email="bob@example.com")
    session.add(user)
    session.commit()

10. 示例完整代码

以下是一个完整的示例，包含定义模型、创建表、CRUD 操作和关系：

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, sessionmaker, relationship

# 创建 Engine
engine = create_engine("sqlite:///example.db", echo=True)

# 创建基类
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String, unique=True)
    posts = relationship("Post", back_populates="user")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="posts")


# 创建会话
Session = sessionmaker(bind=engine)

# CRUD 示例
with Session() as session:
    # 创建用户
    new_user = User(name="Alice", email="alice@example.com")
    session.add(new_user)
    session.commit()

    # 创建帖子
    new_post = Post(title="Hello World", user=new_user)
    session.add(new_post)
    session.commit()

    # 查询用户及其帖子
    user = session.query(User).options(joinedload(User.posts)).first()
    print(user.name, [post.title for post in user.posts])