Peewee ORM 详解：接口使用与实践

Peewee 是一个小型、富有表现力、功能齐全的 Python ORM (Object-Relational Mapper)。它提供了一种简单且 Pythonic 的方式来与数据库进行交互，支持 SQLite、PostgreSQL 和 MySQL 等多种关系型数据库。Peewee 的设计理念是轻量级和易用性，使得开发者可以快速地构建应用程序，而无需编写大量的 SQL 语句。

核心思想：将数据库表映射为 Python 类，将表的行映射为类的实例，将表的列映射为类的属性。 通过 Python 对象和方法来操作数据库，从而抽象掉底层的 SQL 细节。

---

一、为什么选择 Peewee？

在 Python 生态中，存在多种 ORM 解决方案，如 SQLAlchemy、Django ORM 等。Peewee 在其中脱颖而出，主要归因于以下特点：

1. 轻量级与简洁性：Peewee 本身代码量较少，API 设计简洁直观，学习曲线平缓。
2. 富有表现力：其查询 API 允许开发者使用类似 Python 原生语法的方式链式调用，构建复杂的查询。
3. 兼容性强：支持 SQLite、PostgreSQL 和 MySQL 等主流关系型数据库，通过安装对应的驱动即可切换。
4. 灵活性：虽然是 ORM，但不完全限制对 SQL 的使用，允许执行原始 SQL 语句。
5. 活跃开发：项目维护者积极响应问题和功能请求。

二、关键概念

在使用 Peewee 之前，理解几个核心概念至关重要。

1. ORM (Object-Relational Mapper)
   • 定义：ORM 是一种编程技术，用于在面向对象编程语言和关系型数据库之间转换数据。它允许开发者使用对象的概念来操作数据库，而不是直接编写 SQL 语句。ORM 充当了对象模型和关系模型之间的桥梁。
2. Database (数据库)
   • 定义：在 Peewee 中，Database 类（如 SqliteDatabase, PostgresqlDatabase, MySQLDatabase）代表一个特定的数据库连接。所有模型都必须绑定到一个数据库实例才能执行操作。
3. Model (模型)
   • 定义：Model 类是 Peewee 的核心。每个 Model 子类都映射到数据库中的一张表。Model 定义了表的结构，包括字段、关系和元信息。
4. Field (字段)
   • 定义：Field 类（如 CharField, IntegerField, DateTimeField）映射到数据库表中的列。每个字段定义了列的数据类型、约束（如主键、非空、默认值、唯一性）以及其他属性。
5. Query (查询)
   • 定义：Peewee 提供了强大的查询 API，用于构建 SELECT、INSERT、UPDATE、DELETE 等 SQL 操作。这些查询通常通过 Model 类的方法或 SelectQuery 对象链式调用来完成。
6. Transaction (事务)
   • 定义：事务是一组原子操作，要么全部成功提交，要么全部失败回滚。Peewee 提供了 atomic() 上下文管理器来简化事务管理。

三、数据库设置与连接

Peewee 支持多种数据库。首先需要安装相应的 Python 驱动（例如，SQLite 无需额外安装，PostgreSQL 需要 psycopg2，MySQL 需要 PyMySQL 或 mysqlclient）。

3.1 引入数据库模块

from peewee import *

# 针对不同数据库类型，引入对应的数据库类
# SQLite
# db = SqliteDatabase('my_database.db')

# PostgreSQL
# db = PostgresqlDatabase('my_database', user='my_user', password='my_password', host='localhost', port=5432)

# MySQL
# db = MySQLDatabase('my_database', user='my_user', password='my_password', host='localhost', port=3306)

# 以 SQLite 为例进行演示
db = SqliteDatabase('example.db')

3.2 连接与关闭

数据库连接一般通过 db.connect() 建立，并在操作完成后通过 db.close() 关闭。推荐使用 with 语句进行连接管理，确保连接正确关闭。

# 手动连接与关闭
try:
    db.connect()
    # 数据库操作
finally:
    db.close()

# 使用上下文管理器 (推荐)
with db:
    # 数据库操作
    print("数据库连接已建立并将在退出with块后自动关闭。")

四、定义模型 (Model)

模型是 Peewee 中映射数据库表的核心。每个模型类都应继承自 peewee.Model，并定义对应的字段。

4.1 基本模型定义

class BaseModel(Model):
    class Meta:
        database = db # 将模型绑定到之前定义的数据库实例

class User(BaseModel):
    # Field definitions -- start
    id = AutoField(primary_key=True) # 主键，自增
    username = CharField(max_length=50, unique=True, index=True) # 字符串，最大长度50，唯一，上有索引
    email = CharField(max_length=100, unique=True, null=False) # 非空，唯一
    password_hash = CharField(max_length=128)
    is_active = BooleanField(default=True) # 布尔类型，默认True
    created_at = DateTimeField(default=datetime.datetime.now) # 日期时间类型，默认当前时间
    # Field definitions -- end

    class Meta:
        table_name = 'users' # 可选，指定表名，默认为类名小写
        order_by = ('username',) # 可选，查询时默认排序

class Post(BaseModel):
    title = CharField(max_length=255)
    content = TextField() # 文本类型，无长度限制
    author = ForeignKeyField(User, backref='posts') # 外键，关联 User 模型，通过 user.posts 反向查询
    published_at = DateTimeField(null=True)
    is_published = BooleanField(default=False)

关键字段类型与选项：

• AutoField: 主键，自动增长。
• CharField(max_length): 字符串类型，必须指定最大长度。
• TextField(): 长文本类型，无长度限制。
• IntegerField(): 整数类型。
• BigIntegerField(): 大整数类型。
• FloatField(): 浮点数类型。
• BooleanField(): 布尔类型。
• DateTimeField()/DateField()/TimeField(): 日期/时间类型。
• DecimalField(max_digits, decimal_places): 定点数类型，用于精确的货币等。
• BlobField(): 二进制数据类型。
• ForeignKeyField(Model, backref=...): 外键，backref 用于反向查询。

常见字段选项：

• primary_key=True: 设为表主键。
• unique=True: 确保列值唯一。
• null=True/null=False: 允许/不允许 NULL 值。
• default=...: 列的默认值。
• index=True: 为列创建索引，加快查询速度。
• verbose_name="...": 字段的描述性名称（仅在一些高级ORM中用于表单/管理界面）。

4.2 定义关系 (Relationships)

Peewee 通过 ForeignKeyField 定义模型之间的关系。

一对多 (One-to-Many)：
在 Post 模型中的 author = ForeignKeyField(User, backref='posts') 定义了 Post 与 User 之间的一对多关系：一个用户可以有多篇文章，一篇文章只属于一个用户。
backref='posts' 允许我们通过 user_instance.posts 访问该用户的所有文章。

class Comment(BaseModel):
    text = TextField()
    user = ForeignKeyField(User, backref='comments') # 评论属于哪个用户
    post = ForeignKeyField(Post, backref='comments') # 评论属于哪篇文章
    created_at = DateTimeField(default=datetime.datetime.now)

现在，我们有一个用户、文章和评论的简单 ERD (Entity-Relationship Diagram)：

五、数据库操作 (CRUD)

5.1 创建表

在模型定义完成后，需要将它们同步到数据库中，创建实际的表。
注意： db.create_tables() 只会创建不存在的表。如果表已存在，它不会做任何操作。

with db:
    db.create_tables([User, Post, Comment])
    print("表已创建或已存在。")

5.2 增加数据 (Create)

5.2.1 Model.create()

这是最常见的创建新对象的方式。

with db:
    # 创建用户
    try:
        alice = User.create(username='alice', email='alice@example.com', password_hash='hash1')
        bob = User.create(username='bob', email='bob@example.com', password_hash='hash2')
        charlie = User.create(username='charlie', email='charlie@example.com', password_hash='hash3')
        print(f"用户 {alice.username}, {bob.username}, {charlie.username} 已创建。")
    except IntegrityError as e:
        print(f"创建用户失败 (可能已存在): {e}")

    # 创建文章
    try:
        post1 = Post.create(title='Peewee入门', content='这是一篇关于Peewee的文章。', author=alice, is_published=True)
        post2 = Post.create(title='Peewee高级特性', content='本文探讨Peewee的一些高级用法。', author=alice, is_published=False)
        post3 = Post.create(title='我的第一篇文章', content='Hello World!', author=bob, is_published=True)
        print(f"文章 {post1.title}, {post2.title}, {post3.title} 已创建。")
    except IntegrityError as e:
        print(f"创建文章失败 (可能已存在): {e}")

    # 创建评论
    try:
        Comment.create(text='很棒的入门教程！', user=bob, post=post1)
        Comment.create(text='期待高级特性。', user=charlie, post=post1)
        Comment.create(text='学习了！', user=alice, post=post3)
        print("评论已创建。")
    except IntegrityError as e:
        print(f"创建评论失败 (可能已存在): {e}")

5.2.2 实例创建和 save()

可以先创建模型实例，设置属性，再通过 save() 存储到数据库。

with db:
    eve = User(username='eve', email='eve@example.com', password_hash='hash_eve')
    eve.save() # 插入新记录
    print(f"用户 {eve.username} 已通过实例方法创建。")

    # 如果是更新现有记录，也使用 save()
    eve.email = 'new_eve@example.com'
    eve.save() # 更新现有记录
    print(f"用户 {eve.username} 的邮箱已更新。")

5.2.3 bulk_create() / insert_many()

用于批量插入数据，效率更高。

with db:
    users_to_add = [
        {'username': 'frank', 'email': 'frank@example.com', 'password_hash': 'h1'},
        {'username': 'grace', 'email': 'grace@example.com', 'password_hash': 'h2'}
    ]
    # Peewee 3.x 使用 insert_many
    User.insert_many(users_to_add).execute()
    print("Frank 和 Grace 已批量创建。")

5.3 查询数据 (Read)

查询是 ORM 最常用和最重要的功能之一。

5.3.1 Model.select()

select() 方法返回一个 SelectQuery 对象，可以用于构建复杂的查询。

with db:
    # 查询所有用户
    print("\n--- 所有用户 ---")
    for user in User.select():
        print(f"ID: {user.id}, Username: {user.username}, Email: {user.email}, Active: {user.is_active}")

    # 查询特定用户 (使用 .get() 方法，如果未找到会抛出 DoesNotExist 异常)
    print("\n--- 特定用户 (alice) ---")
    try:
        alice = User.get(User.username == 'alice')
        print(f"找到了 Alice: ID={alice.id}, Email={alice.email}")
    except User.DoesNotExist:
        print("未找到 Alice。")

    # 查询特定用户 (使用 .get_or_none() 方法，未找到返回 None)
    print("\n--- 特定用户 (not_exist_user) ---")
    not_exist_user = User.get_or_none(User.username == 'not_exist_user')
    if not_exist_user:
        print(f"找到了 Not Exist User: {not_exist_user.username}")
    else:
        print("未找到 Not Exist User。")

    # 查询所有已发布的文章
    print("\n--- 所有已发布的文章 ---")
    published_posts = Post.select().where(Post.is_published == True)
    for post in published_posts:
        print(f"Title: {post.title}, Author: {post.author.username}") # .author 是一个 User 对象

    # 查询 Alice 发布的所有文章
    print("\n--- Alice 发布的所有文章 ---")
    alice_posts = Post.select().join(User).where(User.username == 'alice')
    # 或者直接使用 alice 实例
    # alice = User.get(User.username == 'alice')
    # alice_posts = Post.select().where(Post.author == alice)
    for post in alice_posts:
        print(f"Title: {post.title}, Content: {post.content[:30]}...")

    # 查询包含特定关键词的文章，并按发布时间倒序
    print("\n--- 包含'Peewee'关键词的文章，按发布时间倒序 ---")
    peewee_posts = Post.select().where(Post.title.contains('Peewee')).order_by(Post.published_at.desc())
    for post in peewee_posts:
        print(f"Title: {post.title}, Published At: {post.published_at}")

    # 限制和偏移 (Limit & Offset)
    print("\n--- 最新的2篇文章 ---")
    latest_posts = Post.select().order_by(Post.created_at.desc()).limit(2)
    for post in latest_posts:
        print(f"Title: {post.title}")

    # 聚合查询 (Count)
    print(f"\n--- 文章总数: {Post.select().count()} ---")
    print(f"Alice 发布的文章数量: {Post.select().where(Post.author.username == 'alice').count()}")

    # 聚合查询 (Average)
    # 假设我们有一个 NumericField，这里使用一个不存在的字段作为示例
    # avg_value = Post.select(fn.AVG(Post.views_count)).scalar()
    # print(f"平均浏览量: {avg_value}")

    # 使用 `fn` (Functions)
    print("\n--- 按用户统计文章数量 ---")
    posts_by_author = (User
                       .select(User.username, fn.COUNT(Post.id).alias('post_count'))
                       .join(Post, JOIN.LEFT_OUTER) # 使用左连接，即使没有文章的用户也会被统计
                       .group_by(User.username)
                       .order_by(fn.COUNT(Post.id).desc()))
    for item in posts_by_author:
        print(f"用户: {item.username}, 文章数量: {item.post_count}")

    # 查询所有评论，并同时获取评论的用户和文章信息 (N+1 问题优化)
    print("\n--- 查询所有评论及关联的用户和文章 ---")
    # `prefetch` 会执行两次查询：一次获取Comment，一次获取关联的User和Post
    for comment in Comment.select().prefetch(User, Post):
        print(f"评论: '{comment.text}' by {comment.user.username} on '{comment.post.title}'")

    # 指定 select 哪些列
    print("\n--- 仅查询用户名和邮箱 ---")
    for user_data in User.select(User.username, User.email):
        print(f"Username: {user_data.username}, Email: {user_data.email}")

5.3.2 复杂的 where 条件

使用 & (AND) 和 | (OR) 操作符来组合条件。

with db:
    # 查询 Alice 或 Bob 发布且已发布的文章
    print("\n--- Alice 或 Bob 发布且已发布的文章 ---")
    users = (User.username == 'alice') | (User.username == 'bob')
    published_by_alice_or_bob = Post.select().join(User).where(users & (Post.is_published == True))
    for post in published_by_alice_or_bob:
        print(f"Title: {post.title}, Author: {post.author.username}")

    # 使用 `in_` 和 `not_in`
    print("\n--- 用户名在列表中的用户 ---")
    users_in_list = User.select().where(User.username.in_(['alice', 'charlie']))
    for user in users_in_list:
        print(f"Username: {user.username}")

5.3.3 原始 SQL

当 Peewee 的查询 API 无法满足需求时，可以使用原始 SQL。

with db:
    print("\n--- 原始 SQL 查询 (所有用户) ---")
    cursor = db.execute_sql('SELECT id, username, email FROM users;')
    for row in cursor.fetchall():
        print(f"ID: {row[0]}, Username: {row[1]}, Email: {row[2]}")

    print("\n--- 原始 SQL 查询 (带参数) ---")
    # 参数化查询防止 SQL 注入
    username_param = 'bob'
    cursor = db.execute_sql('SELECT * FROM users WHERE username = ?;', (username_param,))
    bob_user_raw = cursor.fetchone()
    if bob_user_raw:
        print(f"Raw query found Bob: {bob_user_raw}")

    # 通过 Model.raw() 返回 Model 实例
    print("\n--- Model.raw() 查询 ---")
    raw_query_users = User.raw('SELECT * FROM users WHERE is_active = ?', True)
    for user in raw_query_users:
        print(f"Raw query model: {user.username}")

5.4 更新数据 (Update)

5.4.1 更新单个实例

with db:
    # 获取需要更新的实例
    try:
        alice = User.get(User.username == 'alice')
        print(f"\n--- 更新 Alice 的邮箱 ---")
        old_email = alice.email
        alice.email = 'alice_new@example.com' # 修改属性
        alice.save() # 保存更改到数据库
        print(f"Alice 的邮箱从 {old_email} 更新为 {alice.email}")
    except User.DoesNotExist:
        print("Alice 不存在，无法更新。")

5.4.2 批量更新

使用 Model.update().where() 进行批量更新。

with db:
    # 启用所有非活跃用户
    print("\n--- 批量更新：启用所有非活跃用户 ---")
    inactive_count = User.select().where(User.is_active == False).count()
    if inactive_count > 0:
        updated_count = User.update(is_active=True).where(User.is_active == False).execute()
        print(f"更新了 {updated_count} 个非活跃用户，将其设为活跃。")
    else:
        print("没有非活跃用户需要更新。")

    # 批量发布 Bob 的文章
    print("\n--- 批量发布 Bob 的文章 ---")
    updated_posts_count = Post.update(is_published=True, published_at=datetime.datetime.now()).where(Post.author.username == 'bob').execute()
    print(f"发布了 {updated_posts_count} 篇 Bob 的文章。")

5.5 删除数据 (Delete)

5.5.1 删除单个实例

with db:
    # 删除 Grace 用户
    try:
        grace = User.get(User.username == 'grace')
        print(f"\n--- 删除 Grace 用户 ---")
        grace.delete_instance() # 删除实例
        print(f"用户 {grace.username} 已删除。")
    except User.DoesNotExist:
        print("Grace 不存在，无法删除。")

5.5.2 批量删除

使用 Model.delete().where() 进行批量删除。

with db:
    # 删除所有未发布的文章
    print("\n--- 批量删除：删除所有未发布的文章 ---")
    deleted_count = Post.delete().where(Post.is_published == False).execute()
    print(f"删除了 {deleted_count} 篇未发布的文章。")

注意：外键约束 (Foreign Key Constraints) 可能会影响删除操作。默认情况下，如果存在依赖该记录的外键时进行删除，数据库会阻止删除操作。可以通过在 ForeignKeyField 中设置 on_delete='CASCADE' 来实现级联删除。

六、事务 (Transactions)

事务确保一组数据库操作的原子性。如果其中任何一个操作失败，整个事务都会回滚，所有更改都不会被保存。

Peewee 提供了 db.atomic() 上下文管理器来简化事务处理。

with db:
    # 模拟一个事务，其中一个操作会失败
    try:
        with db.atomic(): # 开启一个事务
            print("\n--- 事务开始 ---")
            # 这是一个成功的操作
            User.create(username='transaction_user_1', email='t1@example.com', password_hash='t1')
            print("创建了 transaction_user_1")

            # 这是一个会失败的操作 (attempt to create a user with existing unique username 'alice')
            User.create(username='alice', email='t2@example.com', password_hash='t2')
            print("尝试创建 transaction_user_2 (应该失败)")

            # 如果上面语句成功，才会执行到这里
            User.create(username='transaction_user_3', email='t3@example.com', password_hash='t3')
            print("创建了 transaction_user_3")

            print("--- 事务提交 (如果所有操作成功) ---")
    except IntegrityError as e:
        print(f"--- 事务回滚！发生错误: {e} ---")
    except Exception as e:
        print(f"--- 事务回滚！发生未知错误: {e} ---")

with db:
    # 验证事务回滚：确保 transaction_user_1 和 transaction_user_3 不存在
    t1 = User.get_or_none(User.username == 'transaction_user_1')
    t3 = User.get_or_none(User.username == 'transaction_user_3')
    print(f"User transaction_user_1 存在: {bool(t1)}")
    print(f"User transaction_user_3 存在: {bool(t3)}")

运行上述代码，transaction_user_1 和 transaction_user_3 都不会被创建，因为 User.create(username='alice', ...) 违反了唯一性约束，导致事务回滚。

七、高级特性

7.1 数据库代理 (DatabaseProxy)

在某些场景下，你可能无法在定义模型时立即提供数据库实例（例如，在应用启动时动态配置数据库）。DatabaseProxy 允许你先定义模型，之后再将实际的数据库实例绑定到代理。

# 1. 创建一个数据库代理
proxy_db = DatabaseProxy()

class ProxyBaseModel(Model):
    class Meta:
        database = proxy_db # 模型绑定到代理

class UserProxy(ProxyBaseModel):
    name = CharField()

# 此时，UserProxy 无法进行数据库操作
# UserProxy.create_table() 会报错

# 2. 之后，将实际的数据库绑定到代理
actual_db = SqliteDatabase(':memory:') # 使用内存数据库
proxy_db.init(actual_db) # 绑定实际数据库实例

# 3. 现在，UserProxy 可以进行数据库操作了
with actual_db:
    actual_db.create_tables([UserProxy])
    UserProxy.create(name='Proxy User')
    print(f"通过代理创建的用户: {UserProxy.get(name='Proxy User').name}")

7.2 迁移 (Migrations)

Peewee 本身没有像 Django 或 Alembic 那样内置的高级数据库迁移工具。对于复杂的生产环境，通常需要结合第三方库或手动管理迁移。

• 手动管理：通过编写 SQL 脚本或使用 Model.add_column(), Model.drop_column(), db.execute_sql() 等方法来手动进行 schema 变更。
• 第三方工具：可以使用 peewee_migrations 等社区项目来辅助管理。

7.3 连接池 (Connection Pooling)

对于需要频繁打开/关闭数据库连接的生产环境，使用连接池可以提高性能。Peewee 的数据库类支持连接池。

# PostgreSQL 示例, min_connections 和 max_connections 是关键参数
# db = PostgresqlDatabase(
#     'my_database',
#     user='my_user',
#     password='my_password',
#     host='localhost',
#     max_connections=20, # 最大连接数
#     stale_timeout=300 # 连接在300秒不活跃后关闭
# )

# SqliteDatabase 也有 pool_size 和 thread_safe 等参数
# db = SqliteDatabase('my_database.db', pragmas={'journal_mode': 'wal'}, autoconnect=True, check_same_thread=False)

八、优缺点与适用场景

8.1 优点：

1. 轻量与快速：代码库小，启动速度快，对资源占用低。
2. API 简洁直观：学习成本低，非常适合小型项目或对 ORM 复杂性有要求的场景。
3. 表达力强：链式调用和 Pythonic 的 API 使得查询构建自然。
4. 性能良好：虽然是 ORM，但在许多情况下能够生成高效的 SQL。
5. 灵活性高：支持混合使用 ORM 查询和原始 SQL。

8.2 缺点：

1. 功能相对精简：相较于 SQLAlchemy 这种全功能 ORM，Peewee 在一些高级功能（如声明式混入、复杂的会话管理、详细的事件系统）上可能有所欠缺。
2. 社区规模较小：相对于 Django 或 SQLAlchemy，Peewee 的用户社区和第三方工具生态可能不那么庞大。
3. 缺乏内置迁移工具：需要手动编写或结合第三方工具进行数据库 schema 迁移。

8.3 适用场景：

• 小型到中型 Web 应用：快速原型开发，RESTful API 后端。
• 命令行工具和脚本：需要快速与数据库交互的自动化脚本。
• 微服务：作为轻量级的数据访问层。
• 教育和学习：作为入门 ORM 的优秀选择。
• 对性能和资源占用有较高要求，但又不想牺牲开发效率的场景。

九、总结

Peewee 是一个出色的 Python ORM，以其简洁、轻量和富有表现力的特点，在 Python 数据库操作中占据一席之地。通过本文详细讲解的数据库设置、模型定义、CRUD 操作及事务管理等接口使用方式，开发者可以高效地利用 Peewee 构建健壮且易于维护的应用程序。虽然它在某些高级功能上可能不如更重量级的 ORM 完善，但其简洁的 API 和优雅的设计使其成为许多 Python 项目的理想选择。在实际开发中，根据项目需求权衡其优缺点，结合最佳实践，将能充分发挥 Peewee 的强大能力。