【超详细】Python ORM 入门指北 —— 用 Peewee 写增删改查

---

1 为什么选择 Peewee？

• 代码极简，上手只需 5 行
• 支持 SQLite / MySQL / PostgreSQL 一键切换
• 单文件 5 k 行源码，阅读无压力
• 与 Flask / FastAPI / Scrapy 都能愉快玩耍

---

2 环境 30 秒搭好

pip install peewee
# 可选：MySQL / PostgreSQL 驱动
# pip install pymysql psycopg2-binary

---

3 建库 & 建表

# db.py
from peewee import *

# 使用 SQLite 文件，生产可换成 MySQLDatabase
db = SqliteDatabase('demo.db')

class BaseModel(Model):
    class Meta:
        database = db

class User(BaseModel):
    id = AutoField(primary_key=True)
    name = CharField(max_length=20, null=False)
    age = IntegerField(default=18)
    email = CharField(unique=True)

# 建表（不存在才创建）
db.connect()
db.create_tables([User])

---

4 增：Create

方式	特点
create()	一行写完，自动保存
先 new 再 save	方便中间处理逻辑

# 方式 A：一步到位
u1 = User.create(name='Alice', age=22, email='alice@example.com')

# 方式 B：先实例化再 save
u2 = User(name='Bob', age=30, email='bob@example.com')
u2.save()          # 手动落库

---

5 查：Retrieve

# 5.1 主键查询
user = User.get_by_id(1)

# 5.2 条件查询：单个
alice = User.get(User.name == 'Alice')

# 5.3 条件查询：多个
adults = User.select().where(User.age >= 18)

# 5.4 排序 + 切片
top3 = User.select().order_by(User.age.desc()).limit(3)

# 5.5 模糊搜索
gmail_users = User.select().where(User.email.endswith('@gmail.com'))

---

6 改：Update

场景	推荐写法
已拿到对象	修改字段 → save()
批量更新	update().execute()

# 单条更新
alice = User.get(User.name == 'Alice')
alice.age = 23
alice.save()

# 批量更新：把所有 18 岁用户改成 19
rows = User.update(age=19).where(User.age == 18).execute()
print(f"更新了 {rows} 行")

---

7 删：Delete

# 单条删除
alice.delete_instance()

# 级联批量删除
rows = User.delete().where(User.age < 18).execute()
print(f"删除了 {rows} 行")

---

8 事务 & 异常处理

from peewee import IntegrityError

try:
    with db.atomic():
        User.create(email='dup@example.com')
        User.create(email='dup@example.com')  # 会触发唯一约束
except IntegrityError as e:
    print('捕获重复邮箱异常：', e)

---

9 完整代码一键跑

# run.py
from db import User, db

def main():
    db.connect()
    db.create_tables([User])

    # 增
    User.create(name='Alice', age=22, email='alice@example.com')
    User.create(name='Bob', age=30, email='bob@example.com')

    # 查
    for u in User.select():
        print(u.id, u.name, u.age)

    # 改
    User.update(age=23).where(User.name == 'Alice').execute()

    # 删
    User.delete().where(User.name == 'Bob').execute()

if __name__ == '__main__':
    main()

---

10 常见报错速查表

报错	原因 & 解决
UserDoesNotExist	get() 没查到记录，用 .get_or_none() 或 try/except
IntegrityError	唯一索引冲突，先查再插或捕获异常
OperationalError: no such table	忘记 create_tables()

---

11 彩蛋：用 ModelSelect 玩链式

query = (User
         .select()
         .where(User.age.between(20, 30))
         .order_by(User.name)
         .paginate(2, 10))  # 第 2 页，每页 10 条

---

12 小结思维导图

Peewee CRUD
├── Create
│   ├── create()
│   └── save()
├── Retrieve
│   ├── get() / get_by_id()
│   ├── select() + where
│   └── order_by / limit / paginate
├── Update
│   ├── 改属性 + save()
│   └── update().execute()
└── Delete
    ├── delete_instance()
    └── delete().execute()

---

如果本篇文章帮到了你，欢迎点赞 + 评论 + 收藏！
更多 Python 实战干货，关注作者不迷路～