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

Park Lam

9 小时前

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自动化工具。