SQLAlchemy

SQLAlchemy是 Python 下的 SQL Toolkit and Object Relational Mapper 框架。

安装

SQLAlchemy 2.0 支持 Python3.7 及以上版本:

Bash

pip install SQLAlchemy

# 预发布版本
pip install --pre SQLAlchemy

SQLAlchemy 底层由特定数据库实现了 Python Database API Specification v2.0 规范的驱动来提供支持，因此还需要安装对应的数据库驱动，下面是常用的数据库的一些驱动:

psycopg2: SQLAlchemy 默认的 PostgreSQL 驱动
sqlite3: Python 标准库提供的 SQLite 驱动，同样是 SQLAlchemy 默认的 SQLite 驱动
PyMySQL: 用于 MySQL/MariaDB 数据库
duckdb-engine: SQLAlchemy 的 DuckDB 驱动

Tips

更多的可以查看的Dialects，其中还包含一些第三方提供的驱动。

架构

SQLAlchemy 有三种不同的 API 来进行操作，可以单独使用或组合在一起使用。其中 SQLAlchemy ORM 是构建在 SQLAlchemy Core 之上的，而 SQLAlchemy Core 又是构建在 DBAPI 之上的。他们由以下组件构成:

SQLAlchemy Arch

DBAPI

他由实现了 Python Database API Specification v2.0 规范的三方库提供，是真正与数据库之间的通信。

Tips

一些特殊的功能通过sqlalchemy.dialects.[driver]暴露出来。

Tips

SQLAlchemy Core 中的 Engine 是对这些驱动的进一步封装，Engine 管理由驱动提供的 Connection 对象并提供了连接池的功能。

SQLAlchemy Core

SQLAlchemy Core 是 SQLAlchemy 的基础架构。提供了用于管理数据库连接、构造 SQL 语句以及与数据库查询结果进行交互的工具:

Schema: 定义数据库架构，通过指定元数据来构造数据库中表、列等架构实例
SQL Expression Language: Core 的核心，用于构造 SQL 语句
Engine: 任何 SQLAlchemy 应用程序的起点，管理数据库连接。底层由 DBAPI 来与数据库进行通信

他们之间是相辅相成的关系，SQL Expression Language 是其中的核心，他使用 Schema 定义来构造 SQL 语句，最终构造的 SQL 语句自动转换为字符串形式传递给 Engine 并由他管的连接来执行。

SQLAlchemy ORM

SQLAlchemy ORM 是构建在 SQLAlchemy Core 之上的，他提供了Mapped Class来使用 Python 中的类来定义数据库模式。并且提供了Session对象持久化机制来操作数据库。

Note

Core 和 SQL Expression Language 提供了以 Schema 为中心的数据库视图以及面向命令(不变性)的编程范式。ORM 在此基础上构建了以 domain 为中心的数据库视图以及面向对象并依赖可变性(状态的变化)的编程范式。

总结

SQLAlchemy 是 Python 下的 ORM 框架，提供了三种操作流程:

DBAPI 流程: 创建 Connection → 使用 Connection.cursor() 返回 Cursor → 编写 SQL 字符串 → 执行 cursor.execute(sql_string) -> 返回结果 -> 关闭 Connection
Core 流程: 编写Schema → 构建SQL Expression Language → 从Engine中获取 Connection → 执行 Connection.execute(sql) → 返回Results → 操作完毕释放 Connection
ORM 流程: 编写Mapped Class → 操作Session → 自动转换为 Core 流程并且更新 Session 状态

可以看到 ORM 的流程要简单的多，我们只需要编写好 Mapped Class 然后操作 Session 即可，之后的流程大部分都是自动完成的。但是 ORM 流程是构建在 Core 之上的，实际上编写 Mapped Class 就是将编写 Schema 换一种写法(ORM 以类构建映射，Core 以方法创建 Schema 对象)。

而 Core 是构建在 DBAPI 之上的，她最核心的地方就是提供了 SQL Expression Language 来避免用户直接手写 SQL 字符串:

Python

# DBAPI
# 创建 Connection 对象
connection = connect(URI)
# 获取 Curosr 对象
cursor = connection.cursor()
# 执行 SQL 语句
result = cursor.execute("SELECT name,fullname WHERE id=1")
connection.close()

# Core
from sqlalchemy import MetaData,create_engine
from sqlalchemy import Table, Column, Integer, String

# SQLAlchemy 应用的起点
engine = create_engine(URI)
# MetaData 即 数据库
metadata_obj = MetaData()

# 通过 Table 方法来构造表对象
# 通过 Column 方法来构造列对象
user = Table(
    "user",
    metadata_obj,
    Column("id", Integer, primary_key=True),
    Column("name", String(16), nullable=False),
    Column("fullname", String(30)),
    Column("nickname", String(50), nullable=True),
)
# 等价于 SELECT name,fullname FROM User WHERE id=1
# select() 方法构造 select 对象，where() 方法构造 where 对象
# 组合 select 对象和 where 对象来构造 SQL 语句
stem = select(User.name, User.fullname).where(User.id == 1)

# 执行构造好的对象来与数据库通信
with engine.connection() as conn:
  conn.execute(stem)

# ORM
from sqlalchemy import Integer, String, ForeignKey,create_engine
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

# SQLAlchemy 应用的起点
engine = create_engine(URI)
# 相当于构建 MetaData 即数据库
class Base(DeclarativeBase):
    pass

# 构建表映射
# 其中的每个属性对应表中的行
class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    fullname: Mapped[str] = mapped_column(String(30))
    nickname: Mapped[Optional[str]]

# 直接操作 Session 对象，状态会自动维护
with Session(engine) as session:
    # 构建对象，对应数据库表中的行, User 是编写的对象模型
    squidward = User(name="squidward", fullname="Squidward Tentacles")
    # IdentitySet([User(id=None, name='squidward', fullname='Squidward Tentacles')])    此时 id = None
    krabs = User(name="ehkrabs", fullname="Eugene H. Krabs")
    # 添加到 session 中
    session.add(squidward)
    session.add(krabs)
    # 刷新状态到数据库中，此时 SQLAlchemy ORM 会自动构建 Insert 语句
    session.flush()
    # 此时数据状态变为 persistent 即持久态
    print(squidward.id)     # 4 可以看到插入数据的同时更新的 session 的数据状态
    # 获取 User 中 id=1 的数据
    # 同样会有 SQLAlchemy ORM 自动构造 Select 语句来执行
    # 获取的数据直接就是持久态的
    session.get(User, 1)
    # 其他的增删改查都是一个道理

Note

SQLAlchemy Core 的 API 通常位于 sqlalchemy 命名空间中，而 SQLAlchemy ORM 位于 sqlalchemy.orm 命名空间中。DBAPI 则位于 sqlalchemy.dialects 命名空间中。