Peewee 轻量 ORM：简洁的 Python 数据库库

FreeGuideOnline 最新 2026-06-16

Peewee 轻量 ORM：简洁的 Python 数据库库

Peewee 是一个简单小巧的 Python ORM（对象关系映射）库。它支持 SQLite、MySQL、PostgreSQL 和 CockroachDB，以直观的 API 和极低的学习成本著称。本教程将带你从安装开始，逐步掌握模型定义、增删改查、查询过滤、事务操作等核心技能。

为什么要选择 Peewee

轻量级：源代码不过数千行，依赖极少，启动快
直觉化的 API：用 Python 代码表达 SQL，却不隐藏 SQL 的强大
开箱即用的支持：内置连接池、迁移工具、信号机制
完整的文档与社区：适合个人项目、中小型应用快速开发

环境安装

使用 pip 安装基础包，如需特定数据库驱动请一并安装：

pip install peewee

数据库驱动的选择：

SQLite：自带 sqlite3 模块，无需额外安装
MySQL：pip install pymysql
PostgreSQL：pip install psycopg2-binary

连接到数据库

Peewee 通过数据库引擎类来管理连接。下面是最常见的数据库连接方式。

from peewee import *

# SQLite（文件数据库）
db = SqliteDatabase('my_app.db')

# MySQL（需先安装 PyMySQL）
db = MySQLDatabase(
    'my_database',
    user='root',
    password='secret',
    host='localhost',
    port=3306
)

# PostgreSQL
db = PostgresqlDatabase(
    'my_db',
    user='postgres',
    password='secret',
    host='127.0.0.1'
)

定义数据模型

模型映射数据库表，每个模型类对应一张表，类属性对应字段。

class BaseModel(Model):
    """所有模型的公共基类"""
    class Meta:
        database = db  # 绑定数据库

class User(BaseModel):
    username = CharField(unique=True, max_length=50)
    email = CharField(max_length=255)
    age = IntegerField(null=True)          # 允许为空
    joined_at = DateTimeField(default=datetime.datetime.now)

    class Meta:
        table_name = 'users'               # 自定义表名
        indexes = (
            (('username', 'email'), True), # 联合唯一索引
        )

常用字段类型：
CharField、TextField、IntegerField、FloatField、DecimalField、BooleanField、DateField、DateTimeField、ForeignKeyField 等。

创建数据表

定义完模型后调用 create_tables() 方法生成对应的数据库表结构。

db.connect()
db.create_tables([User])   # 可传入多个模型
db.close()

也可以使用模型的快捷方法：

User.create_table()

核心操作：增删改查

Peewee 提供了语法糖和原汁原味的 SQL 查询方式，下面逐一介绍。

创建记录 (C)

# 方式一：使用 create() 类方法
user = User.create(username='alice', email='alice@example.com', age=25)

# 方式二：实例化然后 save()
user = User(username='bob', email='bob@example.com', age=30)
user.save()  # 返回影响行数

# 批量插入（性能高）
data = [
    {'username': 'charlie', 'email': 'charlie@x.com'},
    {'username': 'diana', 'email': 'diana@x.com'}
]
User.insert_many(data).execute()

查询记录 (R)

# 获取单条记录
user = User.get(User.username == 'alice')

# 获取多条记录（返回可迭代的 Select 查询）
all_users = User.select()

# 条件过滤：年龄大于等于 25，且用户名以 'a' 开头
query = User.select().where(
    (User.age >= 25) & (User.username.startswith('a'))
)

# 排序与限制
users = (User
         .select()
         .order_by(User.age.desc())
         .limit(10))

# 查询特定字段（避免加载不需要的列）
titles = User.select(User.username, User.email).dicts()
# 转换为字典列表方便序列化

更新记录 (U)

# 更新单个实例
user = User.get(User.username == 'alice')
user.age = 26
user.save()

# 批量更新（返回影响行数）
q = User.update(age=User.age + 1).where(User.username == 'bob')
q.execute()

删除记录 (D)

# 删除单个实例
user = User.get(User.username == 'charlie')
user.delete_instance()  # 级联删除关联对象（如有外键配置）

# 批量删除
User.delete().where(User.age < 20).execute()

高级查询与表达式

Peewee 使用 Python 的运算符重载来构建查询表达式，可读性很强。

# 组合条件：AND 用 &，OR 用 |
adults = User.select().where(
    (User.age >= 18) & ((User.username == 'alice') | (User.username == 'bob'))
)

# 范围查询
young = User.select().where(User.age.between(18, 35))

# IN 查询
names = ['alice', 'bob', 'eve']
User.select().where(User.username.in_(names))

# 聚合函数
from peewee import fn
avg_age = User.select(fn.AVG(User.age)).scalar()
count_per_age = User.select(User.age, fn.COUNT(User.id)).group_by(User.age)

关系与外键

定义一对多、多对多关系是 ORM 的核心能力。

class Tweet(BaseModel):
    user = ForeignKeyField(User, backref='tweets')
    content = TextField()
    created_at = DateTimeField(default=datetime.datetime.now)

# 通过 user 反向获取所有 tweet
user = User.get(User.username == 'alice')
for tweet in user.tweets:   # backref 属性
    print(tweet.content)

# 通过 tweet 正向访问用户
tweet = Tweet.get(Tweet.id == 1)
print(tweet.user.username)

# 多对多关系示例（中间表）
class Student(BaseModel):
    name = CharField()

class Course(BaseModel):
    title = CharField()

class Enrollment(BaseModel):
    student = ForeignKeyField(Student)
    course = ForeignKeyField(Course)

事务处理

确保多个操作要么全部完成，要么全部回滚。

# 上下文管理器方式（推荐）
with db.atomic() as txn:
    user = User.create(username='new_user', email='new@x.com')
    Tweet.create(user=user, content='Hello world!')
    # 如果中间出现异常，自动回滚

手动事务控制：

db.begin()
try:
    # 一系列操作
    db.commit()
except Exception:
    db.rollback()
    raise
finally:
    db.close()

原始 SQL 与 Peewee 混合使用

Peewee 不强制隐藏 SQL，你可以直接执行原生语句或混合使用。

# 执行原生查询
query = "SELECT * FROM users WHERE age > %s"
users = User.raw(query, 20)
for u in users:
    print(u.username)

# 从查询对象获取 SQL 和参数（调试用）
sq = User.select().where(User.age > 20)
print(sq.sql())         # 打印生成的 SQL
print(sq.sql()[0])      # 参数化的 SQL 语句
print(sq.sql()[1])      # 参数元组

连接池与生产配置

对于 Web 应用，推荐使用连接池来减少连接开销。

from playhouse.pool import PooledPostgresqlDatabase

db = PooledPostgresqlDatabase(
    'my_db',
    max_connections=32,
    stale_timeout=300,
    user='postgres',
    password='secret'
)

Playhouse 扩展还提供了 PooledMySQLDatabase 等池化实现。

迁移与 schema 变更

Peewee 内置 playhouse.migrate 用于零停机更新表结构。

from playhouse.migrate import migrate, SqliteMigrator

migrator = SqliteMigrator(db)
migrate(
    migrator.add_column('users', 'bio', TextField(null=True)),
    migrator.rename_column('users', 'email', 'email_address'),
    migrator.add_index('users', ('username',), unique=True)
)

实用技巧与常见陷阱

避免 N+1 查询：使用 select_related() 或 prefetch() 预加载关联对象
分页：利用 .paginate(page_number, page_size) 快速分页
时间处理：字段默认值不要用 datetime.datetime.now，应使用 datetime.datetime.now（无括号）或 DateTimeField(default=datetime.datetime.now)，但如果调用时就会固定为模块加载时间。正确做法是 default=datetime.datetime.now 传递函数引用。
调试模式：import peewee; peewee.DEBUG = True 可打印所有 SQL 语句

# 预加载关联数据避免 N+1
tweets = (Tweet
          .select(Tweet, User)
          .join(User)
          .where(Tweet.created_at > some_date))
for tweet in tweets:
    print(tweet.user.username)   # 不产生额外查询

下一步学习

探索 Playhouse 扩展：信号、加密字段、数据库 URL 连接
结合 FastAPI/Flask 构建 Web 应用
阅读官方文档：docs.peewee-orm.com
测试驱动开发：使用内存 SQLite 编写单元测试

Peewee 用最小的学习代价换来了灵活的数据操作能力。当你想要一个“不喧宾夺主”的 ORM 时，它会是非常体面的选择。