好，各位观众，欢迎来到今天的“SQLAlchemy Core API：构建 SQL 表达式与数据库抽象层”讲座！今天咱们不讲玄学，只讲干货，带大家一起玩转 SQLAlchemy Core，看看它是如何让咱们跟数据库“谈恋爱”的。
开场白：为啥要学 SQLAlchemy Core？
想象一下，你是个厨师，想做一道美味的菜肴。你有两种选择：
- 直接用手抓: 直接写 SQL 语句，像
SELECT * FROM users WHERE age > 25。这很灵活，但也很容易出错，而且代码可读性差。 - 用厨房神器: 使用 SQLAlchemy Core，它可以帮你把复杂的 SQL 操作变成 Python 对象，让你像搭积木一样构建 SQL 语句，减少出错的概率，提高代码可读性。
SQLAlchemy Core 就是那个“厨房神器”。它提供了一套强大的 API，让你可以在 Python 中构建 SQL 表达式，并且可以方便地与各种数据库进行交互。
第一部分： SQLAlchemy Core 的核心概念
SQLAlchemy Core 就像一个乐高玩具，由几个核心部件组成：
- Engine: 数据库连接引擎，负责建立和管理与数据库的连接。
- MetaData: 数据库元数据，描述数据库的结构，例如表、列、索引等。
- Table: 表对象，代表数据库中的一个表。
- Column: 列对象，代表表中的一列。
- SQL 表达式: 使用 Python 对象构建的 SQL 语句，例如
select(),insert(),update(),delete()等。 - Connection: 数据库连接对象，用于执行 SQL 语句。
1. Engine：连接数据库的桥梁
Engine 是 SQLAlchemy 的入口点，它负责建立与数据库的连接。你可以把它想象成一个水龙头，拧开它，水（数据）就哗啦啦地流出来了。
from sqlalchemy import create_engine
# 创建一个 SQLite 数据库引擎 (内存数据库)
engine = create_engine('sqlite:///:memory:')
# 创建一个 PostgreSQL 数据库引擎
# engine = create_engine('postgresql://user:password@host:port/database')
# 创建一个 MySQL 数据库引擎
# engine = create_engine('mysql+pymysql://user:password@host:port/database')这里 create_engine() 函数接收一个数据库 URL 作为参数。URL 的格式取决于你使用的数据库类型。
2. MetaData：数据库的蓝图
MetaData 对象用于存储数据库的元数据，包括表、列、索引等信息。你可以把它想象成数据库的蓝图，告诉你数据库里有什么东西。
from sqlalchemy import MetaData
metadata = MetaData()3. Table：表的化身
Table 对象代表数据库中的一个表。你需要定义表的名称、列以及其他约束条件。你可以把它想象成数据库里的一张桌子，上面放着各种东西（数据）。
from sqlalchemy import Table, Column, Integer, String, Identity
users_table = Table(
'users', metadata,
Column('id', Integer, Identity(), primary_key=True),
Column('name', String(50), nullable=False),
Column('age', Integer)
)这里我们创建了一个名为 users 的表，它有三个列：id (主键，自动增长)、name (字符串，不能为空) 和 age (整数)。
4. Column：列的代言人
Column 对象代表表中的一列。你需要指定列的名称、数据类型以及其他约束条件。你可以把它想象成桌子上的一个抽屉，里面放着特定类型的东西（数据）。
在上面的例子中，Column('id', Integer, Identity(), primary_key=True) 就定义了一个名为 id 的列，它的数据类型是整数，并且是主键，自动增长。
5. SQL 表达式：构建 SQL 语句的积木
SQLAlchemy Core 提供了一系列函数，用于构建各种 SQL 语句，例如 select(), insert(), update(), delete() 等。你可以把它们想象成积木，通过组合这些积木，你可以构建出各种复杂的 SQL 语句。
from sqlalchemy import select, insert, update, delete
# 查询所有用户
select_stmt = select(users_table)
# 插入一个新用户
insert_stmt = insert(users_table).values(name='Alice', age=30)
# 更新用户的年龄
update_stmt = update(users_table).where(users_table.c.name == 'Alice').values(age=31)
# 删除用户
delete_stmt = delete(users_table).where(users_table.c.name == 'Alice')6. Connection：执行 SQL 语句的通道
Connection 对象用于执行 SQL 语句。你可以把它想象成水管，连接水龙头（Engine）和目的地（数据库）。
with engine.connect() as connection:
# 执行查询语句
result = connection.execute(select_stmt)
for row in result:
print(row)
# 执行插入语句
connection.execute(insert_stmt)
# 执行更新语句
connection.execute(update_stmt)
# 执行删除语句
connection.execute(delete_stmt)
connection.commit() # 提交事务第二部分：实战演练：CRUD 操作
现在，让我们通过一个实际的例子来演示如何使用 SQLAlchemy Core 进行 CRUD (Create, Read, Update, Delete) 操作。
1. 创建表
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Identity
engine = create_engine('sqlite:///:memory:')
metadata = MetaData()
users_table = Table(
'users', metadata,
Column('id', Integer, Identity(), primary_key=True),
Column('name', String(50), nullable=False),
Column('age', Integer)
)
metadata.create_all(engine) # 创建所有表这里我们创建了一个名为 users 的表，并使用 metadata.create_all(engine) 创建了所有表。
2. 插入数据 (Create)
from sqlalchemy import insert
with engine.connect() as connection:
insert_stmt = insert(users_table).values([
{'name': 'Alice', 'age': 30},
{'name': 'Bob', 'age': 25},
{'name': 'Charlie', 'age': 35}
])
result = connection.execute(insert_stmt)
connection.commit()
print(result.inserted_primary_key) # 打印插入数据的id这里我们使用 insert() 函数插入了三条数据。
3. 查询数据 (Read)
from sqlalchemy import select
with engine.connect() as connection:
select_stmt = select(users_table)
result = connection.execute(select_stmt)
for row in result:
print(row)这里我们使用 select() 函数查询了所有用户，并打印了结果。
from sqlalchemy import select, and_
with engine.connect() as connection:
select_stmt = select(users_table).where(and_(users_table.c.age > 25, users_table.c.name.like('%li%'))) # 模糊查询包含li的年龄大于25的人
result = connection.execute(select_stmt)
for row in result:
print(row)这里我们使用 select() 函数查询了年龄大于25，并且名字包含 ‘li’ 的用户，并打印了结果。
4. 更新数据 (Update)
from sqlalchemy import update
with engine.connect() as connection:
update_stmt = update(users_table).where(users_table.c.name == 'Alice').values(age=31)
result = connection.execute(update_stmt)
connection.commit()
print(result.rowcount) # 打印更新行数这里我们使用 update() 函数将 Alice 的年龄更新为 31。
5. 删除数据 (Delete)
from sqlalchemy import delete
with engine.connect() as connection:
delete_stmt = delete(users_table).where(users_table.c.name == 'Bob')
result = connection.execute(delete_stmt)
connection.commit()
print(result.rowcount) # 打印删除行数这里我们使用 delete() 函数删除了名为 Bob 的用户。
第三部分：高级用法：连接、函数、表达式
SQLAlchemy Core 的强大之处在于它的灵活性和可扩展性。你可以使用它进行各种高级操作，例如连接、函数和复杂表达式。
1. 连接 (Join)
假设我们有两个表：users 和 addresses。
from sqlalchemy import Table, Column, Integer, String, ForeignKey
addresses_table = Table(
'addresses', metadata,
Column('id', Integer, Identity(), primary_key=True),
Column('user_id', Integer, ForeignKey('users.id')),
Column('email_address', String(100), nullable=False)
)
metadata.create_all(engine)现在我们可以使用 join() 函数将这两个表连接起来。
from sqlalchemy import select, join
with engine.connect() as connection:
join_stmt = select(users_table.c.name, addresses_table.c.email_address).join(addresses_table, users_table.c.id == addresses_table.c.user_id)
result = connection.execute(join_stmt)
for row in result:
print(row)2. 函数 (Functions)
SQLAlchemy Core 提供了许多内置的函数，例如 func.count(), func.sum(), func.avg() 等。你也可以自定义函数。
from sqlalchemy import func
with engine.connect() as connection:
count_stmt = select(func.count()).select_from(users_table) # 查询用户数量
result = connection.execute(count_stmt)
print(result.scalar()) # 获取单个值3. 复杂表达式 (Complex Expressions)
你可以使用 Python 的运算符和逻辑运算符来构建复杂的 SQL 表达式。
from sqlalchemy import and_, or_
with engine.connect() as connection:
select_stmt = select(users_table).where(or_(users_table.c.age > 30, and_(users_table.c.name.like('%A%'), users_table.c.age < 26)))
result = connection.execute(select_stmt)
for row in result:
print(row)第四部分：事务控制
事务是数据库操作的重要组成部分。SQLAlchemy Core 提供了事务控制的功能，确保数据的一致性和完整性。
with engine.connect() as connection:
trans = connection.begin() # 开始事务
try:
insert_stmt = insert(users_table).values(name='David', age=40)
connection.execute(insert_stmt)
update_stmt = update(users_table).where(users_table.c.name == 'Alice').values(age=32)
connection.execute(update_stmt)
trans.commit() # 提交事务
except Exception as e:
trans.rollback() # 回滚事务
print(f"发生错误: {e}")第五部分：绑定参数
为了防止 SQL 注入，你应该使用绑定参数来传递数据。
with engine.connect() as connection:
select_stmt = select(users_table).where(users_table.c.name == :name)
result = connection.execute(select_stmt, {"name": "Alice"}) # 使用字典传递参数
for row in result:
print(row)或者：
with engine.connect() as connection:
select_stmt = select(users_table).where(users_table.c.name == bindparam('name'))
result = connection.execute(select_stmt, name = "Alice") # 使用bindparam
for row in result:
print(row)第六部分：使用反射读取数据库结构
如果你已经有一个现成的数据库，你可以使用反射功能来读取数据库的结构。
from sqlalchemy import create_engine, MetaData, Table
engine = create_engine('sqlite:///existing_database.db') # 修改为你的数据库
metadata = MetaData()
metadata.reflect(bind=engine) # 反射数据库结构
users_table = metadata.tables['users'] # 获取 users 表
print(users_table.columns.keys()) # 打印 users 表的列名第七部分：总结与展望
SQLAlchemy Core 是一个强大的工具，它可以让你在 Python 中构建 SQL 表达式，并方便地与各种数据库进行交互。虽然它比 SQLAlchemy ORM 更底层，但它也更加灵活和可控。
| 特性 | SQLAlchemy Core | SQLAlchemy ORM |
|---|---|---|
| 抽象级别 | 低 | 高 |
| 灵活性 | 高 | 较低 |
| 学习曲线 | 陡峭 | 相对平缓 |
| 适用场景 | 需要精细控制 SQL 语句的复杂应用 | 简单的 CRUD 操作，快速开发 |
| 代码量 | 较多 | 较少 |
| 数据库抽象 | 数据库结构抽象，SQL 表达式生成 | 对象关系映射，数据库表映射到 Python 类 |
| 性能 | 通常更高，因为可以优化 SQL 语句 | 可能会有性能损耗，取决于 ORM 的使用方式 |
希望今天的讲座能够帮助你更好地理解 SQLAlchemy Core，并在实际项目中应用它。记住，熟能生巧，多练习才能真正掌握它！
好了，今天的讲座就到这里，谢谢大家！ 散会，吃饭去！