跳至主要內容

Peewee轻量级ORM实战

Li JiaXin教程后端框架测试开发PeeweeORM大约 13 分钟

Peewee轻量级ORM实战

如果说SQLAlchemy是ORM界的"瑞士军刀"，那么Peewee就是"精致的手术刀"——小巧、精准、易用。作为一名测试开发工程师，我发现Peewee特别适合快速构建测试数据管理系统，它的Django风格API让人感觉非常亲切。

一、Peewee简介：为什么选择Peewee？

ORM框架对比

在Python的ORM生态中，主要有三大选择：

框架优点缺点适用场景
Django ORM易用，学习曲线短，与Django紧密集成QuerySet性能一般，依赖DjangoDjango项目
SQLAlchemy功能强大，灵活设计，适合复杂查询学习曲线陡峭，概念复杂大型复杂项目
Peewee轻量级，Django风格API，易于集成不支持自动schema迁移小型项目，快速开发

Peewee的特点

  • 轻量级：核心代码简洁，依赖少
  • Django风格：API设计类似Django ORM，上手容易
  • 灵活集成：可以与任何Web框架集成
  • 功能完整：支持关系查询、事务、连接池等

二、快速入门

1. 安装Peewee

# 基础安装
pip install peewee

# 如果需要MySQL支持
pip install pymysql

# 如果需要PostgreSQL支持
pip install psycopg2-binary

2. 基础概念

在Peewee中，核心概念的对应关系：

Peewee概念数据库概念说明
Model类数据表定义表结构和行为
Field实例表字段定义字段类型和约束
Model实例表记录具体的数据行

3. 第一个Peewee应用

import datetime
from peewee import *

# 创建数据库连接
db = SqliteDatabase('test_platform.db')

class BaseModel(Model):
    """基础模型类"""
    class Meta:
        database = db

class User(BaseModel):
    """用户模型"""
    username = CharField(unique=True, max_length=50)
    email = CharField(max_length=100)
    password = CharField(max_length=100)
    is_active = BooleanField(default=True)
    created_at = DateTimeField(default=datetime.datetime.now)

class TestCase(BaseModel):
    """测试用例模型"""
    name = CharField(max_length=200)
    description = TextField(null=True)
    author = ForeignKeyField(User, backref='testcases')
    priority = IntegerField(default=1)  # 1-5优先级
    status = CharField(max_length=20, default='active')
    created_at = DateTimeField(default=datetime.datetime.now)
    updated_at = DateTimeField(default=datetime.datetime.now)

# 创建表
if __name__ == '__main__':
    db.connect()
    db.create_tables([User, TestCase])
    print("数据表创建成功！")

三、字段类型详解

常用字段类型

from peewee import *
import datetime

class TestModel(BaseModel):
    # 字符串字段
    name = CharField(max_length=100)                    # 短字符串
    description = TextField()                           # 长文本
    
    # 数值字段
    age = IntegerField()                               # 整数
    score = FloatField()                               # 浮点数
    price = DecimalField(max_digits=10, decimal_places=2)  # 精确小数
    
    # 布尔字段
    is_active = BooleanField(default=True)             # 布尔值
    
    # 日期时间字段
    created_at = DateTimeField(default=datetime.datetime.now)  # 日期时间
    birth_date = DateField()                           # 日期
    login_time = TimeField()                           # 时间
    
    # 特殊字段
    data = BlobField()                                 # 二进制数据
    config = JSONField()                               # JSON数据（需要数据库支持）

# 字段约束
class UserProfile(BaseModel):
    username = CharField(unique=True, max_length=50)   # 唯一约束
    email = CharField(index=True)                      # 索引
    age = IntegerField(null=True)                      # 允许为空
    status = CharField(choices=[('active', '激活'), ('inactive', '禁用')])
    created_at = DateTimeField(default=datetime.datetime.now, index=True)

关系字段

class Department(BaseModel):
    """部门模型"""
    name = CharField(max_length=100)
    code = CharField(max_length=20, unique=True)

class Employee(BaseModel):
    """员工模型"""
    name = CharField(max_length=50)
    email = CharField(max_length=100)
    # 外键关系：多对一
    department = ForeignKeyField(Department, backref='employees')

class Project(BaseModel):
    """项目模型"""
    name = CharField(max_length=100)
    description = TextField()

class ProjectMember(BaseModel):
    """项目成员模型（多对多关系的中间表）"""
    employee = ForeignKeyField(Employee)
    project = ForeignKeyField(Project)
    role = CharField(max_length=50, default='member')
    joined_at = DateTimeField(default=datetime.datetime.now)

# 使用示例
if __name__ == '__main__':
    # 创建部门
    dev_dept = Department.create(name='开发部', code='DEV')
    
    # 创建员工
    employee = Employee.create(
        name='张三',
        email='zhangsan@example.com',
        department=dev_dept
    )
    
    # 查询部门下的所有员工
    for emp in dev_dept.employees:
        print(f"员工: {emp.name}, 邮箱: {emp.email}")

四、CRUD操作详解

1. 创建数据（Create）

# 方法一：使用create()
user = User.create(
    username='testuser',
    email='test@example.com',
    password='123456'
)

# 方法二：实例化后保存
user = User(
    username='testuser2',
    email='test2@example.com',
    password='123456'
)
user.save()

# 方法三：批量创建
users_data = [
    {'username': 'user1', 'email': 'user1@example.com', 'password': '123'},
    {'username': 'user2', 'email': 'user2@example.com', 'password': '123'},
    {'username': 'user3', 'email': 'user3@example.com', 'password': '123'},
]

# 批量插入（高效）
User.insert_many(users_data).execute()

# 获取或创建
user, created = User.get_or_create(
    username='admin',
    defaults={'email': 'admin@example.com', 'password': 'admin123'}
)
if created:
    print("创建了新用户")
else:
    print("用户已存在")

2. 查询数据（Read）

# 基础查询
# 获取所有用户
users = User.select()
for user in users:
    print(f"用户: {user.username}, 邮箱: {user.email}")

# 条件查询
active_users = User.select().where(User.is_active == True)

# 获取单个记录
try:
    user = User.get(User.username == 'admin')
    print(f"找到用户: {user.username}")
except User.DoesNotExist:
    print("用户不存在")

# 复杂查询条件
users = User.select().where(
    (User.is_active == True) & 
    (User.created_at > datetime.datetime(2024, 1, 1))
)

# 模糊查询
users = User.select().where(User.username.contains('test'))
users = User.select().where(User.email.startswith('admin'))

# 排序
users = User.select().order_by(User.created_at.desc())

# 分页
page = 1
page_size = 10
users = User.select().paginate(page, page_size)

# 限制数量
recent_users = User.select().order_by(User.created_at.desc()).limit(5)

# 聚合查询
user_count = User.select().count()
active_count = User.select().where(User.is_active == True).count()

# 字段选择（只查询特定字段）
usernames = User.select(User.username, User.email)

3. 更新数据（Update）

# 更新单个记录
user = User.get(User.username == 'testuser')
user.email = 'newemail@example.com'
user.save()

# 批量更新
User.update(is_active=False).where(
    User.created_at < datetime.datetime(2023, 1, 1)
).execute()

# 原子更新（避免并发问题）
User.update(login_count=User.login_count + 1).where(
    User.username == 'admin'
).execute()

# 条件更新
updated_count = User.update(
    status='inactive'
).where(
    User.is_active == False
).execute()
print(f"更新了 {updated_count} 条记录")

4. 删除数据（Delete）

# 删除单个记录
user = User.get(User.username == 'testuser')
user.delete_instance()

# 批量删除
deleted_count = User.delete().where(
    User.is_active == False
).execute()
print(f"删除了 {deleted_count} 条记录")

# 级联删除（删除用户及其所有测试用例）
user = User.get(User.username == 'testuser')
# 先删除关联的测试用例
TestCase.delete().where(TestCase.author == user).execute()
# 再删除用户
user.delete_instance()

五、关系查询

1. 一对多关系查询

# 查询用户及其测试用例
user = User.get(User.username == 'admin')

# 通过backref访问关联对象
for testcase in user.testcases:
    print(f"测试用例: {testcase.name}")

# 预加载关联对象（避免N+1查询问题）
users_with_testcases = User.select().prefetch(TestCase)
for user in users_with_testcases:
    print(f"用户: {user.username}")
    for testcase in user.testcases:
        print(f"  - 测试用例: {testcase.name}")

# 连接查询
query = (TestCase
         .select(TestCase, User)
         .join(User)
         .where(User.is_active == True))

for testcase in query:
    print(f"测试用例: {testcase.name}, 作者: {testcase.author.username}")

2. 聚合查询

# 统计每个用户的测试用例数量
from peewee import fn

user_stats = (User
              .select(User.username, fn.COUNT(TestCase.id).alias('testcase_count'))
              .join(TestCase, JOIN.LEFT_OUTER)
              .group_by(User.id))

for user in user_stats:
    print(f"用户: {user.username}, 测试用例数: {user.testcase_count}")

# 复杂聚合
stats = (TestCase
         .select(
             TestCase.status,
             fn.COUNT(TestCase.id).alias('count'),
             fn.AVG(TestCase.priority).alias('avg_priority')
         )
         .group_by(TestCase.status))

for stat in stats:
    print(f"状态: {stat.status}, 数量: {stat.count}, 平均优先级: {stat.avg_priority}")

六、实战案例：测试平台数据模型

import datetime
from peewee import *

# 数据库配置
db = SqliteDatabase('test_platform.db')

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

class User(BaseModel):
    """用户模型"""
    username = CharField(unique=True, max_length=50)
    email = CharField(max_length=100)
    password = CharField(max_length=100)
    role = CharField(max_length=20, default='tester')  # admin, tester, developer
    is_active = BooleanField(default=True)
    created_at = DateTimeField(default=datetime.datetime.now)
    last_login = DateTimeField(null=True)

class Project(BaseModel):
    """项目模型"""
    name = CharField(max_length=100)
    description = TextField(null=True)
    owner = ForeignKeyField(User, backref='owned_projects')
    status = CharField(max_length=20, default='active')
    created_at = DateTimeField(default=datetime.datetime.now)

class TestSuite(BaseModel):
    """测试套件模型"""
    name = CharField(max_length=100)
    description = TextField(null=True)
    project = ForeignKeyField(Project, backref='test_suites')
    created_by = ForeignKeyField(User, backref='created_suites')
    created_at = DateTimeField(default=datetime.datetime.now)

class TestCase(BaseModel):
    """测试用例模型"""
    name = CharField(max_length=200)
    description = TextField(null=True)
    test_suite = ForeignKeyField(TestSuite, backref='test_cases')
    author = ForeignKeyField(User, backref='authored_cases')
    priority = IntegerField(default=3)  # 1-5
    status = CharField(max_length=20, default='active')
    steps = JSONField(default=list)  # 测试步骤
    expected_result = TextField(null=True)
    created_at = DateTimeField(default=datetime.datetime.now)
    updated_at = DateTimeField(default=datetime.datetime.now)

class TestExecution(BaseModel):
    """测试执行记录"""
    test_case = ForeignKeyField(TestCase, backref='executions')
    executor = ForeignKeyField(User, backref='executed_tests')
    result = CharField(max_length=20)  # passed, failed, skipped
    duration = FloatField(null=True)  # 执行时间（秒）
    error_message = TextField(null=True)
    executed_at = DateTimeField(default=datetime.datetime.now)

# 数据库操作类
class TestPlatformDB:
    """测试平台数据库操作类"""
    
    @staticmethod
    def init_db():
        """初始化数据库"""
        db.connect()
        db.create_tables([User, Project, TestSuite, TestCase, TestExecution])
        
        # 创建默认管理员用户
        admin, created = User.get_or_create(
            username='admin',
            defaults={
                'email': 'admin@example.com',
                'password': 'admin123',
                'role': 'admin'
            }
        )
        if created:
            print("创建默认管理员用户成功")
    
    @staticmethod
    def create_project(name, description, owner_username):
        """创建项目"""
        try:
            owner = User.get(User.username == owner_username)
            project = Project.create(
                name=name,
                description=description,
                owner=owner
            )
            return project
        except User.DoesNotExist:
            raise ValueError(f"用户 {owner_username} 不存在")
    
    @staticmethod
    def get_project_stats(project_id):
        """获取项目统计信息"""
        try:
            project = Project.get_by_id(project_id)
            
            # 统计测试套件数量
            suite_count = TestSuite.select().where(TestSuite.project == project).count()
            
            # 统计测试用例数量
            case_count = (TestCase
                         .select()
                         .join(TestSuite)
                         .where(TestSuite.project == project)
                         .count())
            
            # 统计最近执行结果
            recent_executions = (TestExecution
                               .select(TestExecution.result, fn.COUNT(TestExecution.id).alias('count'))
                               .join(TestCase)
                               .join(TestSuite)
                               .where(TestSuite.project == project)
                               .where(TestExecution.executed_at > datetime.datetime.now() - datetime.timedelta(days=7))
                               .group_by(TestExecution.result))
            
            execution_stats = {exec.result: exec.count for exec in recent_executions}
            
            return {
                'project_name': project.name,
                'suite_count': suite_count,
                'case_count': case_count,
                'recent_executions': execution_stats
            }
        except Project.DoesNotExist:
            raise ValueError(f"项目 {project_id} 不存在")

# 使用示例
if __name__ == '__main__':
    # 初始化数据库
    TestPlatformDB.init_db()
    
    # 创建项目
    project = TestPlatformDB.create_project(
        name='电商系统测试',
        description='电商系统功能测试项目',
        owner_username='admin'
    )
    
    print(f"项目创建成功: {project.name}")

## 七、高级特性

### 1. 事务处理

```python
from peewee import *

# 方法一：使用装饰器
@db.atomic()
def create_user_with_project(username, email, project_name):
    """原子操作：创建用户和项目"""
    user = User.create(username=username, email=email, password='123456')
    project = Project.create(name=project_name, owner=user)
    return user, project

# 方法二：使用上下文管理器
def transfer_project_ownership(project_id, new_owner_username):
    """转移项目所有权"""
    try:
        with db.atomic():
            project = Project.get_by_id(project_id)
            new_owner = User.get(User.username == new_owner_username)

            # 记录操作日志
            old_owner = project.owner
            project.owner = new_owner
            project.save()

            print(f"项目 {project.name} 从 {old_owner.username} 转移给 {new_owner.username}")

    except (Project.DoesNotExist, User.DoesNotExist) as e:
        print(f"操作失败: {e}")
    except Exception as e:
        print(f"事务回滚: {e}")

# 方法三：手动事务控制
def batch_update_test_results(execution_data):
    """批量更新测试结果"""
    with db.atomic() as transaction:
        try:
            for data in execution_data:
                TestExecution.create(**data)

            # 如果某个条件不满足，可以手动回滚
            if len(execution_data) > 100:
                transaction.rollback()
                print("数据量过大，操作已回滚")
                return False

        except Exception as e:
            print(f"批量更新失败: {e}")
            return False

    return True

2. 数据库连接管理

from peewee import *
import threading

# 连接池配置
from playhouse.pool import PooledMySQLDatabase

# 使用连接池
db = PooledMySQLDatabase(
    'test_platform',
    max_connections=20,
    stale_timeout=300,
    user='root',
    password='password',
    host='localhost',
    port=3306
)

# 多线程安全的数据库操作
class DatabaseManager:
    """数据库管理器"""

    @staticmethod
    def get_connection():
        """获取数据库连接"""
        if db.is_closed():
            db.connect()
        return db

    @staticmethod
    def close_connection():
        """关闭数据库连接"""
        if not db.is_closed():
            db.close()

    @classmethod
    def execute_in_thread(cls, func, *args, **kwargs):
        """在线程中执行数据库操作"""
        def wrapper():
            try:
                cls.get_connection()
                return func(*args, **kwargs)
            finally:
                cls.close_connection()

        thread = threading.Thread(target=wrapper)
        thread.start()
        return thread

# 使用示例
def background_cleanup():
    """后台清理过期数据"""
    cutoff_date = datetime.datetime.now() - datetime.timedelta(days=30)

    # 删除30天前的测试执行记录
    deleted_count = TestExecution.delete().where(
        TestExecution.executed_at < cutoff_date
    ).execute()

    print(f"清理了 {deleted_count} 条过期记录")

# 在后台线程中执行清理
cleanup_thread = DatabaseManager.execute_in_thread(background_cleanup)

3. 查询优化

# 预加载关联对象（解决N+1查询问题）
def get_projects_with_details():
    """获取项目及其详细信息"""
    # 错误的方式（会产生N+1查询）
    # projects = Project.select()
    # for project in projects:
    #     print(f"项目: {project.name}, 所有者: {project.owner.username}")  # 每次都查询数据库

    # 正确的方式（预加载）
    projects = Project.select().join(User).where(Project.status == 'active')

    for project in projects:
        print(f"项目: {project.name}, 所有者: {project.owner.username}")

# 使用prefetch进行复杂预加载
def get_users_with_projects_and_cases():
    """获取用户及其项目和测试用例"""
    users = (User
             .select()
             .where(User.is_active == True)
             .prefetch(Project, TestCase))

    for user in users:
        print(f"用户: {user.username}")
        for project in user.owned_projects:
            print(f"  项目: {project.name}")
            # 通过项目获取测试用例
            cases = TestCase.select().join(TestSuite).where(TestSuite.project == project)
            for case in cases:
                print(f"    测试用例: {case.name}")

# 使用子查询
def get_active_projects_with_recent_activity():
    """获取有最近活动的活跃项目"""
    recent_date = datetime.datetime.now() - datetime.timedelta(days=7)

    # 子查询：最近有执行记录的测试用例
    recent_cases = (TestCase
                   .select(TestCase.id)
                   .join(TestExecution)
                   .where(TestExecution.executed_at > recent_date))

    # 主查询：包含这些测试用例的项目
    active_projects = (Project
                      .select()
                      .join(TestSuite)
                      .join(TestCase)
                      .where(TestCase.id.in_(recent_cases))
                      .distinct())

    return active_projects

# 原生SQL查询（复杂查询时使用）
def get_test_execution_statistics():
    """获取测试执行统计（使用原生SQL）"""
    query = """
    SELECT
        p.name as project_name,
        COUNT(te.id) as total_executions,
        SUM(CASE WHEN te.result = 'passed' THEN 1 ELSE 0 END) as passed_count,
        SUM(CASE WHEN te.result = 'failed' THEN 1 ELSE 0 END) as failed_count,
        AVG(te.duration) as avg_duration
    FROM project p
    JOIN testsuite ts ON ts.project_id = p.id
    JOIN testcase tc ON tc.test_suite_id = ts.id
    JOIN testexecution te ON te.test_case_id = tc.id
    WHERE te.executed_at > date('now', '-30 days')
    GROUP BY p.id, p.name
    ORDER BY total_executions DESC
    """

    cursor = db.execute_sql(query)
    results = []
    for row in cursor.fetchall():
        results.append({
            'project_name': row[0],
            'total_executions': row[1],
            'passed_count': row[2],
            'failed_count': row[3],
            'avg_duration': row[4]
        })

    return results

4. 数据验证和钩子

from peewee import *
import hashlib

class User(BaseModel):
    username = CharField(unique=True, max_length=50)
    email = CharField(max_length=100)
    password = CharField(max_length=100)
    is_active = BooleanField(default=True)
    created_at = DateTimeField(default=datetime.datetime.now)

    def save(self, *args, **kwargs):
        """重写save方法，添加密码加密"""
        # 如果密码被修改，进行加密
        if self.password and not self.password.startswith('hashed_'):
            self.password = f"hashed_{hashlib.md5(self.password.encode()).hexdigest()}"

        super().save(*args, **kwargs)

    @classmethod
    def create_user(cls, username, email, password):
        """创建用户的类方法"""
        # 验证用户名格式
        if len(username) < 3:
            raise ValueError("用户名至少3个字符")

        # 验证邮箱格式
        if '@' not in email:
            raise ValueError("邮箱格式不正确")

        # 验证密码强度
        if len(password) < 6:
            raise ValueError("密码至少6个字符")

        return cls.create(username=username, email=email, password=password)

    def verify_password(self, password):
        """验证密码"""
        hashed_input = f"hashed_{hashlib.md5(password.encode()).hexdigest()}"
        return self.password == hashed_input

class TestCase(BaseModel):
    name = CharField(max_length=200)
    description = TextField(null=True)
    author = ForeignKeyField(User, backref='authored_cases')
    priority = IntegerField(default=3)
    status = CharField(max_length=20, default='active')
    created_at = DateTimeField(default=datetime.datetime.now)
    updated_at = DateTimeField(default=datetime.datetime.now)

    def save(self, *args, **kwargs):
        """重写save方法，自动更新updated_at"""
        self.updated_at = datetime.datetime.now()
        super().save(*args, **kwargs)

    @property
    def priority_text(self):
        """优先级文本描述"""
        priority_map = {
            1: '最低',
            2: '低',
            3: '中',
            4: '高',
            5: '最高'
        }
        return priority_map.get(self.priority, '未知')

    def execute(self, executor, result, duration=None, error_message=None):
        """执行测试用例"""
        execution = TestExecution.create(
            test_case=self,
            executor=executor,
            result=result,
            duration=duration,
            error_message=error_message
        )
        return execution

# 使用示例
if __name__ == '__main__':
    try:
        # 创建用户
        user = User.create_user(
            username='testuser',
            email='test@example.com',
            password='123456'
        )
        print(f"用户创建成功: {user.username}")

        # 验证密码
        if user.verify_password('123456'):
            print("密码验证成功")

        # 创建测试用例
        testcase = TestCase.create(
            name='登录功能测试',
            description='验证用户登录功能是否正常',
            author=user,
            priority=4
        )
        print(f"测试用例创建成功: {testcase.name}, 优先级: {testcase.priority_text}")

        # 执行测试用例
        execution = testcase.execute(
            executor=user,
            result='passed',
            duration=2.5
        )
        print(f"测试执行完成: {execution.result}")

    except ValueError as e:
        print(f"验证错误: {e}")
    except Exception as e:
        print(f"操作失败: {e}")

八、与Flask集成

from flask import Flask, request, jsonify
from peewee import *
import datetime

# Flask应用配置
app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'

# 数据库配置
db = SqliteDatabase('test_platform.db')

# 请求钩子：管理数据库连接
@app.before_request
def before_request():
    if db.is_closed():
        db.connect()

@app.after_request
def after_request(response):
    if not db.is_closed():
        db.close()
    return response

# API路由
@app.route('/api/users', methods=['GET'])
def get_users():
    """获取用户列表"""
    page = int(request.args.get('page', 1))
    size = int(request.args.get('size', 10))

    users = User.select().paginate(page, size)

    return jsonify({
        'code': 200,
        'data': [
            {
                'id': user.id,
                'username': user.username,
                'email': user.email,
                'is_active': user.is_active,
                'created_at': user.created_at.isoformat()
            }
            for user in users
        ]
    })

@app.route('/api/users', methods=['POST'])
def create_user():
    """创建用户"""
    data = request.get_json()

    try:
        user = User.create_user(
            username=data['username'],
            email=data['email'],
            password=data['password']
        )

        return jsonify({
            'code': 201,
            'message': '用户创建成功',
            'data': {
                'id': user.id,
                'username': user.username,
                'email': user.email
            }
        }), 201

    except ValueError as e:
        return jsonify({
            'code': 400,
            'message': str(e)
        }), 400
    except IntegrityError:
        return jsonify({
            'code': 400,
            'message': '用户名已存在'
        }), 400

@app.route('/api/projects/<int:project_id>/stats', methods=['GET'])
def get_project_stats(project_id):
    """获取项目统计信息"""
    try:
        stats = TestPlatformDB.get_project_stats(project_id)
        return jsonify({
            'code': 200,
            'data': stats
        })
    except ValueError as e:
        return jsonify({
            'code': 404,
            'message': str(e)
        }), 404

if __name__ == '__main__':
    # 初始化数据库
    TestPlatformDB.init_db()
    app.run(debug=True)

九、最佳实践

1. 项目结构建议

test_platform/
├── app.py              # Flask应用入口
├── models/             # 数据模型
│   ├── __init__.py
│   ├── base.py         # 基础模型
│   ├── user.py         # 用户模型
│   ├── project.py      # 项目模型
│   └── testcase.py     # 测试用例模型
├── services/           # 业务逻辑
│   ├── __init__.py
│   ├── user_service.py
│   └── project_service.py
├── api/                # API路由
│   ├── __init__.py
│   ├── user_api.py
│   └── project_api.py
├── config.py           # 配置文件
└── requirements.txt    # 依赖列表

2. 性能优化建议

  • 使用连接池：生产环境中使用数据库连接池
  • 预加载关联对象：避免N+1查询问题
  • 合理使用索引：为经常查询的字段添加索引
  • 批量操作：使用insert_many()进行批量插入
  • 查询优化：复杂查询考虑使用原生SQL

3. 常见陷阱

  • 忘记关闭连接：在Web应用中要正确管理数据库连接
  • 事务使用不当：长时间事务会影响性能
  • 字段类型选择：根据实际需求选择合适的字段类型
  • 外键约束：注意外键删除时的级联问题

十、总结

Peewee作为轻量级ORM框架，具有以下优势：

✅ 简单易用 - Django风格API，学习成本低 ✅ 轻量级 - 核心简洁，依赖少 ✅ 功能完整 - 支持关系查询、事务、连接池 ✅ 灵活集成 - 可与任何Web框架集成 ✅ 文档清晰 - 官方文档详细易懂

适用场景

  • 小型项目：快速开发和原型验证
  • 测试工具：测试数据管理和报告系统
  • 学习项目：理解ORM概念和数据库操作
  • 微服务：轻量级服务的数据持久化

🎯 学习建议：从基础的CRUD操作开始，逐步学习关系查询和高级特性。在实际项目中，要注意性能优化和最佳实践的应用。

Peewee虽然轻量，但功能强大，是Python开发者工具箱中不可缺少的一把"利器"！