SQLAlchemy 快速入门：Python ORM 实践指南 – wiki基地

作者： /

SQLAlchemy 快速入门：Python ORM 实践指南

在 Python 的数据库编程世界中，SQLAlchemy 无疑是一个功能强大且广受欢迎的工具。它不仅仅是一个对象关系映射 (ORM) 库，更是一个全面的数据库工具包，能够为各种规模的应用程序提供灵活、高效的数据库抽象。本文将带你快速入门 SQLAlchemy，了解其核心概念和基本实践。

什么是 ORM？为什么选择 SQLAlchemy？

ORM (Object-Relational Mapping) 是一种编程技术，用于在面向对象编程语言和关系型数据库之间转换数据。简单来说，它允许你使用 Python 对象来操作数据库表，而不是直接编写 SQL 语句。这大大提高了开发效率，减少了样板代码，并使数据库操作更加面向对象。

SQLAlchemy 的优势：

  1. 灵活性和控制力： SQLAlchemy 提供了两种主要的工作模式：ORM 和 SQL Expression Language (SQL 表达式语言)。你可以选择完全拥抱 ORM 的便利，也可以在需要时回退到更接近 SQL 的表达式语言，进行细粒度的控制，甚至直接执行原生 SQL。
  2. 高性能： SQLAlchemy 在设计时就考虑到了性能，它的 SQL 编译能力和连接池管理都非常高效。
  3. 数据库兼容性： 支持几乎所有主流的关系型数据库，如 PostgreSQL, MySQL, SQLite, Oracle, SQL Server 等。
  4. 强大的生态系统： 拥有庞大的社区和丰富的文档，遇到问题时很容易找到解决方案。
  5. 会话管理： 引入了“会话”概念，更好地管理数据库事务和对象生命周期。

核心概念

在深入代码之前，我们先了解 SQLAlchemy 的几个核心概念：

  • Engine (引擎): 连接数据库的入口，负责管理数据库连接和方言。
  • Declarative Base (声明式基类): 创建 ORM 模型的基础，我们定义的 Python 类将继承它。
  • Table (表): 映射到数据库中的物理表。ORM 模型会自动创建或与之关联。
  • Column (列): 映射到数据库表中的列，定义了数据类型、约束等。
  • Session (会话): 应用程序与数据库交互的主要接口。它管理着数据库操作的事务，以及加载、创建、修改和删除的对象。
  • Relationship (关系): 定义表之间的关联，如一对一、一对多、多对多。

快速入门实践

我们将使用 SQLite 数据库作为示例，因为它不需要额外的安装配置。

1. 安装 SQLAlchemy

首先，确保你的 Python 环境中安装了 SQLAlchemy：

bash
pip install SQLAlchemy

2. 定义数据库模型

我们来创建一个简单的 User 模型，它将映射到数据库中的 users 表。

“`python
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

1. 创建 Engine

连接到 SQLite 数据库文件 ‘example.db’

echo=True 会打印所有执行的 SQL 语句，方便调试

engine = create_engine(‘sqlite:///example.db’, echo=True)

2. 创建 Declarative Base

所有的 ORM 模型类都将继承这个基类

Base = declarative_base()

3. 定义 User 模型

class User(Base):
tablename = ‘users’ # 映射到数据库中的表名

id = Column(Integer, primary_key=True) # 主键，自动递增
name = Column(String(50), nullable=False) # 姓名，最大长度50，不可为空
email = Column(String(120), unique=True, nullable=False) # 邮箱，唯一且不可为空

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

4. 创建表 (如果不存在)

这会检查 Base 注册的所有模型，并在数据库中创建相应的表

Base.metadata.create_all(engine)

print(“数据库表已创建或已存在。”)
“`

3. 创建 Session

Session 是与数据库进行所有对话的核心。

“`python

5. 创建 Session 工厂

sessionmaker 绑定了 engine，用于创建 Session 实例

Session = sessionmaker(bind=engine)

6. 创建 Session 实例

session = Session()

print(“Session 已创建。”)
“`

4. 增删改查 (CRUD)

现在我们有了 User 模型和 Session，可以开始进行数据库操作了。

A. 添加数据 (Create)

“`python

创建新的 User 对象

new_user1 = User(name=’Alice’, email=’[email protected]’)
new_user2 = User(name=’Bob’, email=’[email protected]’)

将对象添加到会话

session.add(new_user1)
session.add(new_user2)

提交事务，将数据保存到数据库

session.commit()
print(“用户 Alice 和 Bob 已添加。”)

也可以批量添加

new_users = [
User(name=’Charlie’, email=’[email protected]’),
User(name=’David’, email=’[email protected]’)
]
session.add_all(new_users)
session.commit()
print(“用户 Charlie 和 David 已批量添加。”)
“`

B. 查询数据 (Read)

“`python
from sqlalchemy import select

查询所有用户

使用 select() 函数构建查询语句

users = session.scalars(select(User)).all() # scalars().all() 获取所有结果对象
print(“\n所有用户:”)
for user in users:
print(user)

按 ID 查询单个用户

user_by_id = session.scalar(select(User).filter_by(id=1)) # filter_by() 进行条件过滤
print(f”\nID 为 1 的用户: {user_by_id}”)

按名称模糊查询 (使用 like)

users_like_a = session.scalars(select(User).filter(User.name.like(‘%a%’))).all()
print(“\n名字中包含 ‘a’ 的用户:”)
for user in users_like_a:
print(user)

组合条件查询 (and_)

from sqlalchemy import and_
active_users = session.scalars(select(User).filter(and_(User.id > 1, User.name != ‘Bob’))).all()
print(“\nID 大于 1 且名字不是 Bob 的用户:”)
for user in active_users:
print(user)

排序

sorted_users = session.scalars(select(User).order_by(User.name)).all()
print(“\n按名字排序的用户:”)
for user in sorted_users:
print(user)

限制和偏移

limited_users = session.scalars(select(User).limit(2).offset(1)).all()
print(“\n限制2个，从第1个偏移的用户:”)
for user in limited_users:
print(user)
“`

C. 更新数据 (Update)

“`python

查询要更新的用户

user_to_update = session.scalar(select(User).filter_by(name=’Alice’))
if user_to_update:
user_to_update.email = ‘[email protected]’ # 修改对象属性
session.commit() # 提交会话，保存更改
print(f”\n用户 {user_to_update.name} 的邮箱已更新为 {user_to_update.email}”)
else:
print(“\n未找到用户 Alice。”)
“`

D. 删除数据 (Delete)

“`python

查询要删除的用户

user_to_delete = session.scalar(select(User).filter_by(name=’Bob’))
if user_to_delete:
session.delete(user_to_delete) # 从会话中删除对象
session.commit() # 提交会话，删除数据库中的记录
print(f”\n用户 {user_to_delete.name} 已从数据库中删除。”)
else:
print(“\n未找到用户 Bob。”)

再次查询所有用户，验证删除操作

users_after_delete = session.scalars(select(User)).all()
print(“\n删除 Bob 后的所有用户:”)
for user in users_after_delete:
print(user)
“`

5. 关闭 Session

无论操作成功与否，都应该关闭 Session 以释放资源。

python
session.close()
print("\nSession 已关闭。")

在实际应用中，通常会使用 try...finally 块或上下文管理器来确保 session 被正确关闭。

python
with Session() as session:
try:
# 进行数据库操作
new_user = User(name='Eve', email='[email protected]')
session.add(new_user)
session.commit()
print(f"\n用户 {new_user.name} 已成功添加。")
except Exception as e:
session.rollback() # 发生错误时回滚事务
print(f"操作失败: {e}")

总结

通过这篇快速入门，你应该已经掌握了 SQLAlchemy 的核心概念和基本的 CRUD 操作。SQLAlchemy 的功能远不止于此，例如：

  • 关系映射： 如何定义一对多、多对多关系。
  • 表联结 (Joins)： 如何在查询中联结多张表。
  • 事务管理： 更高级的事务控制。
  • SQL 表达式语言： 当 ORM 不足时，如何直接构建 SQL 语句。
  • 数据迁移： 配合 Alembic 等工具管理数据库 Schema 变更。

SQLAlchemy 是一个深度和广度兼备的库。随着你对 Python 数据库编程需求的增长，深入学习其文档将为你带来巨大的收益。希望这篇指南能为你探索 SQLAlchemy 的奇妙世界提供一个良好的开端！

滚动至顶部