2. 环境搭建与第一个数据访问程序

在上一节中我们已经明确了 SQLAlchemy 在数据访问层中的定位：它既是 ORM 框架，也是独立的数据库工具包。本节要完成的是从零到一的环境搭建，并写出第一个能真正连上数据库、执行 SQL、拿到结果的最小程序。整个过程围绕五个关键动作展开：安装 SQLAlchemy、选择并安装数据库驱动、创建 Engine、执行原生 SQL、理解连接池与方言。这五步不是孤立的操作，而是一条从依赖到运行、从配置到结果的完整链路。

安装 SQLAlchemy 与选择数据库驱动

SQLAlchemy 本身通过 pip 安装即可，它不绑定任何具体数据库。真正与数据库通信的是 DBAPI 驱动，SQLAlchemy 通过方言（Dialect）适配不同驱动。因此安装时要分两步：先装框架，再装驱动。

# 安装 SQLAlchemy 2.x
pip install "SQLAlchemy>=2.0"

# 以 PostgreSQL 为例，安装 psycopg 驱动
pip install "psycopg[binary]"

# 若使用 MySQL，可安装 PyMySQL
pip install PyMySQL

# 若使用 SQLite，Python 标准库自带 sqlite3，无需额外安装

驱动选择需要关注三点：数据库类型、同步还是异步、驱动成熟度。同步场景下，PostgreSQL 推荐 psycopg（psycopg3），MySQL 推荐 PyMySQL 或 mysqlclient，SQLite 直接用内置 sqlite3。异步场景则使用 asyncpg、aiomysql 等，并配合 create_async_engine。驱动选错，最典型的报错就是 ModuleNotFoundError: No module named 'psycopg2' 或 Can't load plugin: sqlalchemy.dialects:postgresql.psycopg2，本质是方言找不到对应驱动。

创建 Engine：连接信息的统一入口

Engine 是 SQLAlchemy 访问数据库的核心对象，它并不等于一个连接，而是连接池与方言的组合管理器。创建 Engine 的标准方式是 create_engine()，参数是一个 URL。URL 的通用格式为：

dialect+driver://username:password@host:port/database

# PostgreSQL 示例
postgresql+psycopg://app_user:secret@127.0.0.1:5432/shop

# MySQL 示例
mysql+pymysql://app_user:secret@127.0.0.1:3306/shop?charset=utf8mb4

# SQLite 文件示例
sqlite:///./shop.db

# SQLite 内存示例
sqlite:///:memory:

URL 中的 dialect+driver 决定了使用哪个方言和驱动。如果只写 postgresql://，SQLAlchemy 会按默认驱动尝试加载，容易因环境差异失败，因此生产代码建议显式写全。创建 Engine 时还可以传入关键参数：echo=True 用于打印生成的 SQL，便于调试；pool_size、max_overflow、pool_pre_ping 用于控制连接池行为；connect_args 用于传递驱动特有的连接参数。

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg://app_user:secret@127.0.0.1:5432/shop",
    echo=True,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
)

执行原生 SQL 并读取结果

Engine 创建后，通过 engine.connect() 获取连接，再用 text() 包装 SQL 语句执行。SQLAlchemy 2.x 要求显式使用 text()，这是为了避免字符串拼接带来的注入风险，也便于绑定参数。

from sqlalchemy import text

with engine.connect() as conn:
    # 建表
    conn.execute(text("""
        CREATE TABLE IF NOT EXISTS product (
            id SERIAL PRIMARY KEY,
            name VARCHAR(64) NOT NULL,
            price NUMERIC(10, 2) NOT NULL
        )
    """))

    # 插入，使用绑定参数
    conn.execute(
        text("INSERT INTO product (name, price) VALUES (:name, :price)"),
        {"name": "机械键盘", "price": 399.00},
    )
    conn.commit()

    # 查询
    rows = conn.execute(text("SELECT id, name, price FROM product ORDER BY id"))
    for row in rows:
        print(row.id, row.name, row.price)

这里有两个容易踩坑的点。第一，SQLAlchemy 2.x 默认不会自动提交，执行 INSERT、UPDATE、DELETE 后必须调用 conn.commit()，否则连接关闭时事务回滚，数据不会落库。第二，查询结果 rows 是 Result 对象，可以迭代，也可以使用 rows.mappings() 得到字典式行，或 rows.scalars() 取单列。若使用 engine.begin() 上下文管理器，则会在退出时自动提交或回滚，更适合写操作。

连接池与方言：Engine 背后的两条主线

连接池解决的是“频繁建立和销毁数据库连接代价高”的问题。Engine 默认使用 QueuePool，维护一定数量的空闲连接，请求到来时复用，超出 pool_size 后允许创建 max_overflow 个临时连接，超过上限则等待。连接不是永久有效的，数据库可能因超时、重启或网络中断而断开，pool_pre_ping=True 会在借出连接前先发一个轻量探测，失效连接被丢弃并重建，避免程序拿到坏连接后报错。

方言则是 SQLAlchemy 与具体数据库之间的翻译层。它负责把统一的 SQLAlchemy 表达式编译成目标数据库能识别的 SQL，并处理类型映射、自增主键、分页语法、标识符引用等差异。例如同样是自增主键，PostgreSQL 用 SERIAL，MySQL 用 AUTO_INCREMENT，SQLite 用 INTEGER PRIMARY KEY AUTOINCREMENT，这些差异都由方言吸收。理解这一点，就能明白为什么换数据库时主要改的是 URL 和驱动，而不是业务代码。

一个可运行的最小程序

把上面的片段组合起来，就是一个完整可运行的程序。下面以 SQLite 为例，因为它无需额外安装驱动，适合第一次验证环境。

from sqlalchemy import create_engine, text

engine = create_engine("sqlite:///./shop.db", echo=True)

with engine.begin() as conn:
    conn.execute(text("""
        CREATE TABLE IF NOT EXISTS product (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name VARCHAR(64) NOT NULL,
            price NUMERIC(10, 2) NOT NULL
        )
    """))
    conn.execute(
        text("INSERT INTO product (name, price) VALUES (:name, :price)"),
        {"name": "机械键盘", "price": 399.00},
    )

with engine.connect() as conn:
    result = conn.execute(text("SELECT id, name, price FROM product"))
    for row in result:
        print(f"id={row.id}, name={row.name}, price={row.price}")

运行后，如果 echo=True 生效，控制台会打印出 SQLAlchemy 生成的 SQL 和参数，这能帮你确认方言是否正确工作。查询输出类似 id=1, name=机械键盘, price=399.00，说明从建表、插入到查询的链路已经打通。若报错，优先检查三处：驱动是否安装、URL 格式是否正确、写操作是否提交。

到这里，环境搭建与第一个数据访问程序已经完成。你应当能够独立完成：安装 SQLAlchemy 与驱动、根据数据库类型写出正确的 URL、创建带连接池参数的 Engine、用 text() 执行原生 SQL 并读取结果，以及解释连接池和方言各自解决什么问题。这些是后续 ORM 映射、会话管理和事务控制的基础，建议在本地至少用 SQLite 和另一种数据库各跑通一次，形成可复用的环境检查习惯。