Python使用工具：peewee库使用教程

作者：

在

1. Python生态中的轻量级ORM工具——peewee简介

Python作为一门多功能的编程语言，凭借其简洁的语法和丰富的库生态，已成为数据科学、Web开发、自动化测试等众多领域的首选工具。在数据库操作领域，尽管SQL语言本身具有强大的表达能力，但直接编写SQL语句不仅繁琐，还容易引发安全隐患。对象关系映射（ORM）技术的出现，为Python开发者提供了一种更自然的方式来操作数据库。

peewee是一个轻量级的Python ORM库，由知名Python开发者Charles Leifer于2010年创建。与Django ORM的”大而全”不同，peewee设计理念更注重简洁性和灵活性，它提供了直观的API，同时支持多种数据库后端，包括SQLite、MySQL和PostgreSQL等。截至2023年，peewee在GitHub上拥有超过6.5k的星标，被广泛应用于中小型项目、脚本工具以及需要快速实现数据库交互的场景中。

2. peewee的技术特点与适用场景

2.1 核心功能与工作原理

peewee的核心功能包括：

定义模型类映射数据库表结构
支持关系型数据库操作（增删改查）
提供丰富的字段类型和验证机制
支持事务处理和数据库迁移
提供表达式语法构建复杂查询

其工作原理基于Python的元类（metaclass）和描述器（descriptor）机制，通过定义继承自Model的类来映射数据库表，类属性映射为表字段。当执行查询时，peewee将Python对象操作转换为SQL语句并执行，最后将查询结果转换回Python对象。

2.2 优缺点分析

优点：

轻量级设计：不依赖其他框架，安装简单（仅需pip install peewee）
灵活的数据库支持：无缝切换不同数据库后端
直观的API：代码可读性高，学习曲线平缓
扩展性强：支持自定义字段类型和数据库操作
良好的文档：官方文档详细且提供丰富示例

缺点：

不适合超大型项目：相比SQLAlchemy，复杂查询支持较弱
迁移工具简单：自带的migrate工具功能有限，复杂迁移需依赖第三方工具
社区资源较少：相较于Django ORM，相关教程和第三方插件较少

2.3 许可证信息

peewee采用MIT许可证，这意味着它可以自由用于商业项目，且无需公开源代码，非常适合各类开源和闭源项目使用。

3. peewee的安装与环境配置

3.1 安装方法

使用pip可以轻松安装peewee：

pip install peewee

如果需要使用数据库迁移功能，可以额外安装playhouse扩展：

pip install peewee[playhouse]

3.2 数据库驱动安装

根据使用的数据库后端不同，需要安装相应的驱动：

SQLite：无需额外安装（Python内置支持）
MySQL：pip install pymysql
PostgreSQL：pip install psycopg2-binary

3.3 环境配置示例

以下是一个简单的环境配置示例，展示如何连接不同类型的数据库：

from peewee import *

# SQLite数据库连接
sqlite_db = SqliteDatabase('my_app.db')

# MySQL数据库连接
mysql_db = MySQLDatabase(
    'my_database',  # 数据库名
    user='root',    # 用户名
    password='password',  # 密码
    host='localhost',     # 主机
    port=3306             # 端口
)

# PostgreSQL数据库连接
postgres_db = PostgresqlDatabase(
    'my_database',
    user='postgres',
    password='password',
    host='localhost',
    port=5432
)

# 选择使用的数据库
database = sqlite_db  # 可根据需要切换

4. peewee基础操作详解

4.1 定义数据模型

使用peewee定义数据模型非常直观，只需创建继承自Model的类，并定义相应的字段：

from peewee import *

# 连接SQLite数据库
db = SqliteDatabase('students.db')

# 定义基类模型
class BaseModel(Model):
    class Meta:
        database = db

# 定义学生模型
class Student(BaseModel):
    name = CharField(max_length=100, null=False)  # 姓名，不能为空
    age = IntegerField()                          # 年龄
    gender = CharField(max_length=10, default='unknown')  # 性别，默认为unknown
    created_at = DateTimeField(default=datetime.datetime.now)  # 创建时间

# 定义课程模型
class Course(BaseModel):
    name = CharField(max_length=100)
    teacher = CharField(max_length=50)

# 定义学生选课关系模型（多对多关系）
class Enrollment(BaseModel):
    student = ForeignKeyField(Student, backref='enrollments')
    course = ForeignKeyField(Course, backref='enrollments')
    grade = FloatField(null=True)  # 成绩，可为空

4.2 数据库操作

4.2.1 创建表

# 创建所有表
db.connect()
db.create_tables([Student, Course, Enrollment])

4.2.2 插入数据

# 插入单个学生
student1 = Student.create(name='张三', age=20, gender='男')

# 批量插入学生
students_data = [
    {'name': '李四', 'age': 21, 'gender': '男'},
    {'name': '王五', 'age': 20, 'gender': '女'}
]
Student.insert_many(students_data).execute()

# 插入课程和选课关系
course1 = Course.create(name='Python编程', teacher='李教授')
course2 = Course.create(name='数据结构', teacher='王教授')

Enrollment.create(student=student1, course=course1, grade=90.5)
Enrollment.create(student=student1, course=course2, grade=88.0)

4.2.3 查询数据

# 查询单个记录
student = Student.get(Student.id == 1)
print(f"学生: {student.name}, 年龄: {student.age}")

# 查询所有学生
students = Student.select()
for student in students:
    print(f"学生: {student.name}, 性别: {student.gender}")

# 条件查询
female_students = Student.select().where(Student.gender == '女')
print(f"女生人数: {female_students.count()}")

# 复杂查询：查找选修Python课程的学生
python_course = Course.get(Course.name == 'Python编程')
python_students = (Student
                   .select()
                   .join(Enrollment)
                   .join(Course)
                   .where(Course.id == python_course.id))

for student in python_students:
    print(f"{student.name} 选修了 {python_course.name}")

# 聚合查询：计算平均年龄
avg_age = Student.select(fn.AVG(Student.age)).scalar()
print(f"学生平均年龄: {avg_age}")

4.2.4 更新数据

# 更新单个记录
student = Student.get(Student.id == 1)
student.age = 22
student.save()  # 保存修改

# 批量更新
Student.update(age=Student.age + 1).where(Student.age &lt; 25).execute()

4.2.5 删除数据

# 删除单个记录
student = Student.get(Student.id == 3)
student.delete_instance()

# 批量删除
Student.delete().where(Student.age > 30).execute()

5. peewee高级特性

5.1 事务处理

# 使用上下文管理器进行事务处理
with db.atomic():
    # 创建学生
    student = Student.create(name='赵六', age=23, gender='男')

    # 创建课程
    course = Course.create(name='机器学习', teacher='张教授')

    # 创建选课记录
    Enrollment.create(student=student, course=course, grade=92.0)

# 手动处理事务
try:
    db.begin()
    # 执行数据库操作
    student = Student.get(Student.id == 1)
    student.age = 25
    student.save()

    # 可能会抛出异常的操作
    if student.age > 24:
        raise ValueError("年龄过大")

    db.commit()
except Exception as e:
    db.rollback()
    print(f"操作失败: {e}")

5.2 数据库迁移

使用peewee的playhouse扩展中的migrate模块可以进行数据库迁移：

from playhouse.migrate import *

# 创建迁移器
migrator = SqliteMigrator(db)

# 定义迁移操作
with db.atomic():
    # 添加字段
    migrate(
        migrator.add_column('student', 'email', CharField(null=True)),
        migrator.add_column('course', 'duration', IntegerField(default=16)),
    )

# 创建自定义迁移脚本
class Migration:
    def migrate(self):
        # 执行迁移操作
        pass

    def rollback(self):
        # 回滚操作
        pass

5.3 自定义字段类型

# 自定义JSON字段类型
class JSONField(TextField):
    def db_value(self, value):
        # 将Python对象转换为JSON字符串
        return json.dumps(value)

    def python_value(self, value):
        # 将JSON字符串转换为Python对象
        if value is not None:
            return json.loads(value)

# 在模型中使用自定义字段
class User(BaseModel):
    name = CharField()
    settings = JSONField(default={})

# 使用示例
user = User.create(name='测试用户', settings={'theme': 'dark', 'language': 'zh-CN'})
print(user.settings['theme'])  # 输出: dark

5.4 数据库连接池

from playhouse.pool import PooledMySQLDatabase

# 创建连接池
db = PooledMySQLDatabase(
    'my_database',
    max_connections=8,  # 最大连接数
    stale_timeout=300,  # 连接超时时间（秒）
    user='root',
    password='password',
    host='localhost',
    port=3306
)

# 在请求处理前连接数据库
def before_request():
    db.connect()

# 在请求处理后关闭数据库连接
def after_request():
    if not db.is_closed():
        db.close()

6. peewee与Web框架集成

6.1 与Flask集成

from flask import Flask
from peewee import *

app = Flask(__name__)
db = SqliteDatabase('flask_app.db')

# 定义模型
class User(Model):
    username = CharField(unique=True)
    email = CharField()

    class Meta:
        database = db

# 初始化数据库
def init_db():
    db.connect()
    db.create_tables([User], safe=True)
    db.close()

# 请求前连接数据库
@app.before_request
def before_request():
    db.connect()

# 请求后关闭数据库
@app.after_request
def after_request(response):
    db.close()
    return response

# 路由示例
@app.route('/')
def index():
    users = User.select()
    return render_template('index.html', users=users)

@app.route('/add_user/&lt;username>/&lt;email>')
def add_user(username, email):
    try:
        User.create(username=username, email=email)
        return f"用户 {username} 添加成功"
    except IntegrityError:
        return f"用户 {username} 已存在"

if __name__ == '__main__':
    init_db()
    app.run(debug=True)

6.2 与FastAPI集成

from fastapi import FastAPI, HTTPException
from peewee import *
from pydantic import BaseModel

app = FastAPI()
db = SqliteDatabase('fastapi_app.db')

# 定义模型
class User(BaseModel):
    username: str
    email: str

class UserModel(Model):
    username = CharField(unique=True)
    email = CharField()

    class Meta:
        database = db

# 初始化数据库
def init_db():
    db.connect()
    db.create_tables([UserModel], safe=True)
    db.close()

# 数据库操作依赖
def get_db():
    db.connect()
    try:
        yield
    finally:
        if not db.is_closed():
            db.close()

# 创建用户
@app.post("/users/")
def create_user(user: User, db=Depends(get_db)):
    try:
        user_model = UserModel.create(
            username=user.username,
            email=user.email
        )
        return {"id": user_model.id, **user.dict()}
    except IntegrityError:
        raise HTTPException(status_code=400, detail="Username already exists")

# 获取用户列表
@app.get("/users/")
def read_users(skip: int = 0, limit: int = 100, db=Depends(get_db)):
    users = list(UserModel.select().offset(skip).limit(limit))
    return [{"id": user.id, "username": user.username, "email": user.email} for user in users]

init_db()

7. 实际案例：博客系统实现

7.1 需求分析

我们将使用peewee实现一个简单的博客系统，包含以下功能：

用户注册与登录
文章发布、编辑和删除
评论功能
标签分类

7.2 数据模型设计

from peewee import *
import datetime

db = SqliteDatabase('blog.db')

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

class User(BaseModel):
    username = CharField(unique=True)
    password = CharField()  # 实际应用中应使用哈希密码
    email = CharField(unique=True)
    created_at = DateTimeField(default=datetime.datetime.now)

class Post(BaseModel):
    title = CharField(max_length=200)
    content = TextField()
    author = ForeignKeyField(User, backref='posts')
    created_at = DateTimeField(index=True, default=datetime.datetime.now)
    updated_at = DateTimeField(default=datetime.datetime.now)
    is_published = BooleanField(default=False)

class Comment(BaseModel):
    content = TextField()
    author = ForeignKeyField(User, backref='comments')
    post = ForeignKeyField(Post, backref='comments')
    created_at = DateTimeField(default=datetime.datetime.now)

class Tag(BaseModel):
    name = CharField(unique=True)

class PostTag(BaseModel):
    post = ForeignKeyField(Post)
    tag = ForeignKeyField(Tag)

    class Meta:
        indexes = (
            (('post', 'tag'), True),  # 确保组合唯一
        )

7.3 核心功能实现

# 用户管理
def create_user(username, email, password):
    try:
        with db.atomic():
            user = User.create(
                username=username,
                email=email,
                password=password  # 实际应用中应进行哈希处理
            )
        return user
    except IntegrityError:
        return None

# 文章管理
def create_post(author, title, content, tags=None, is_published=False):
    with db.atomic():
        post = Post.create(
            author=author,
            title=title,
            content=content,
            is_published=is_published
        )

        if tags:
            for tag_name in tags:
                tag, _ = Tag.get_or_create(name=tag_name)
                PostTag.create(post=post, tag=tag)

        return post

def update_post(post, title=None, content=None, tags=None, is_published=None):
    with db.atomic():
        if title is not None:
            post.title = title
        if content is not None:
            post.content = content
        if is_published is not None:
            post.is_published = is_published

        post.updated_at = datetime.datetime.now()
        post.save()

        if tags is not None:
            # 清除旧标签
            PostTag.delete().where(PostTag.post == post).execute()

            # 添加新标签
            for tag_name in tags:
                tag, _ = Tag.get_or_create(name=tag_name)
                PostTag.create(post=post, tag=tag)

        return post

# 评论管理
def create_comment(author, post, content):
    with db.atomic():
        comment = Comment.create(
            author=author,
            post=post,
            content=content
        )
        return comment

# 查询功能
def get_published_posts(limit=10, offset=0):
    return (Post
            .select(Post, User)
            .join(User)
            .where(Post.is_published == True)
            .order_by(Post.created_at.desc())
            .limit(limit)
            .offset(offset))

def get_post_with_comments(post_id):
    try:
        return (Post
                .select(Post, User, Comment)
                .join(User)
                .switch(Post)
                .join(Comment, JOIN.LEFT_OUTER)
                .where(Post.id == post_id, Post.is_published == True)
                .get())
    except Post.DoesNotExist:
        return None

def get_posts_by_tag(tag_name, limit=10, offset=0):
    return (Post
            .select(Post, User, Tag)
            .join(User)
            .switch(Post)
            .join(PostTag)
            .join(Tag)
            .where(Tag.name == tag_name, Post.is_published == True)
            .order_by(Post.created_at.desc())
            .limit(limit)
            .offset(offset))

7.4 命令行界面实现

import click

@click.group()
def cli():
    pass

@cli.command()
@click.option('--username', prompt=True)
@click.option('--email', prompt=True)
@click.option('--password', prompt=True, hide_input=True, confirmation_prompt=True)
def create_user(username, email, password):
    user = create_user(username, email, password)
    if user:
        click.echo(f"用户 {username} 创建成功 (ID: {user.id})")
    else:
        click.echo("创建失败: 用户名或邮箱已存在")

@cli.command()
@click.option('--user-id', type=int, prompt=True)
@click.option('--title', prompt=True)
@click.option('--content', prompt=True)
@click.option('--tags', prompt='标签 (用逗号分隔)', default='')
@click.option('--published/--draft', default=False)
def create_post(user_id, title, content, tags, published):
    try:
        author = User.get(User.id == user_id)
    except User.DoesNotExist:
        click.echo("错误: 用户不存在")
        return

    tag_list = [tag.strip() for tag in tags.split(',') if tag.strip()]
    post = create_post(author, title, content, tag_list, published)
    click.echo(f"文章 {title} 创建成功 (ID: {post.id})")

@cli.command()
@click.option('--post-id', type=int, prompt=True)
@click.option('--title', default=None)
@click.option('--content', default=None)
@click.option('--tags', default=None)
@click.option('--published/--draft', default=None)
def update_post(post_id, title, content, tags, published):
    try:
        post = Post.get(Post.id == post_id)
    except Post.DoesNotExist:
        click.echo("错误: 文章不存在")
        return

    tag_list = None
    if tags is not None:
        tag_list = [tag.strip() for tag in tags.split(',') if tag.strip()]

    updated_post = update_post(post, title, content, tag_list, published)
    click.echo(f"文章 {updated_post.title} 更新成功")

@cli.command()
@click.option('--limit', type=int, default=10)
@click.option('--offset', type=int, default=0)
def list_posts(limit, offset):
    posts = get_published_posts(limit, offset)
    for post in posts:
        tags = [pt.tag.name for pt in post.posttag_set]
        click.echo(f"{post.id}. {post.title} by {post.author.username} ({', '.join(tags)})")

if __name__ == '__main__':
    db.connect()
    db.create_tables([User, Post, Comment, Tag, PostTag])
    db.close()
    cli()

8. 性能优化与最佳实践

8.1 查询优化

# 避免N+1查询问题
# 不好的写法
for post in Post.select():
    print(post.author.username)  # 每次循环都会执行一次查询

# 好的写法（预加载）
for post in Post.select().join(User).prefetch(User):
    print(post.author.username)  # 只执行两次查询

# 使用批量查询
from peewee import chunked

for batch in chunked(Post.select(), 100):
    for post in batch:
        # 处理每篇文章
        pass

8.2 连接管理

# 使用连接池
from playhouse.pool import PooledSqliteDatabase

db = PooledSqliteDatabase(
    'my_app.db',
    max_connections=8,
    stale_timeout=300
)

# 在Web应用中使用请求上下文管理连接
def before_request():
    db.connect()

def after_request(response):
    db.close()
    return response

8.3 索引优化

class Post(Model):
    title = CharField(max_length=200)
    content = TextField()
    author = ForeignKeyField(User, backref='posts')
    created_at = DateTimeField(index=True)  # 创建索引
    is_published = BooleanField(index=True)  # 创建索引

    class Meta:
        indexes = (
            (('author', 'created_at'), False),  # 组合索引
        )

8.4 事务处理

# 使用事务批量插入数据
with db.atomic():
    for i in range(1000):
        Post.create(
            title=f"文章 {i}",
            content="内容...",
            author=author,
            is_published=True
        )

# 或者使用批量插入
data = [
    {'title': '文章1', 'content': '...', 'author': author},
    {'title': '文章2', 'content': '...', 'author': author},
]

with db.atomic():
    Post.insert_many(data).execute()

9. 相关资源

Pypi地址：https://pypi.org/project/peewee/
Github地址：https://github.com/coleifer/peewee
官方文档地址：https://docs.peewee-orm.com/

通过本文的介绍，我们可以看到peewee作为一个轻量级ORM库，提供了简洁而强大的数据库操作能力。无论是简单的脚本工具，还是复杂的Web应用，peewee都能很好地满足需求。其直观的API设计和灵活的数据库支持，使得开发者可以专注于业务逻辑的实现，而不必过多关注底层数据库操作的细节。

关注我，每天分享一个实用的Python自动化工具。

实用工具

Python使用工具：peewee库使用教程

1. Python生态中的轻量级ORM工具——peewee简介

2. peewee的技术特点与适用场景

2.1 核心功能与工作原理

2.2 优缺点分析

2.3 许可证信息

3. peewee的安装与环境配置

3.1 安装方法

3.2 数据库驱动安装

3.3 环境配置示例

4. peewee基础操作详解

4.1 定义数据模型

4.2 数据库操作

4.2.1 创建表

4.2.2 插入数据

4.2.3 查询数据

4.2.4 更新数据

4.2.5 删除数据

5. peewee高级特性

5.1 事务处理

5.2 数据库迁移

5.3 自定义字段类型

5.4 数据库连接池

6. peewee与Web框架集成

6.1 与Flask集成

6.2 与FastAPI集成

7. 实际案例：博客系统实现

7.1 需求分析

7.2 数据模型设计

7.3 核心功能实现

7.4 命令行界面实现

8. 性能优化与最佳实践

8.1 查询优化

8.2 连接管理

8.3 索引优化

8.4 事务处理

9. 相关资源

更多文章

Python使用工具：peewee库使用教程

Python实用工具：Elasticsearch库详解

Google Cloud Storage Python 库深度使用指南：从基础到实践的全方位解析

Redis：Python开发中的高性能数据缓存与消息队列解决方案