Python:Peewee实践记录

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS PostgreSQL,高可用系列 2核4GB
云数据库 RDS MySQL,高可用系列 2核4GB
简介: Python:Peewee实践记录

文档:http://docs.peewee-orm.com/

安装

$ pip install peewee

将已有数据表转为Model

# 导出数据表为Model
$ python -m pwiz -e mysql -H localhost -p 3306 -u root -P -o -i -t user data > user.py

打印执行SQL

import logging
# 打印日志
logger = logging.getLogger('peewee')
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.DEBUG)
logger.propagate = False  # 不向上传播

modle和dict转换

from playhouse.shortcuts import model_to_dict, dict_to_model
# 快捷方式类
class ShortcutModel(object):
    @classmethod
    def from_dict(cls, data, ignore_unknown=False):
        return dict_to_model(cls, data=data, ignore_unknown=ignore_unknown)
    def to_dict(self,
                recurse=True,
                backrefs=False,
                only=None,
                exclude=None,
                seen=None,
                extra_attrs=None,
                fields_from_query=None,
                max_depth=None,
                manytomany=False):
        return model_to_dict(
            self,
            recurse=recurse,
            backrefs=backrefs,
            only=only,
            exclude=exclude,
            seen=seen,
            extra_attrs=extra_attrs,
            fields_from_query=fields_from_query,
            max_depth=max_depth,
            manytomany=manytomany
        )

示例

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

外键关联操作

1、建表

from datetime import datetime
from peewee import *
db = SqliteDatabase('people.db')
class BaseModel(Model):
    class Meta:
        database = db 
class Person(BaseModel):
    id = IntegerField(primary_key=True)
    name = CharField()
    birthday = DateField()
    class Meta:
        table_name = 'person'
class Pet(BaseModel):
    id = IntegerField(primary_key=True)
    # 一对多: 一个Person -> 多个Pet
    owner = ForeignKeyField(Person, backref='pets')
    name = CharField()
    create_time = DateTimeField(default=datetime.now)
    update_time = DateTimeField(default=datetime.now)
    class Meta:
        table_name = 'pet'
# 创建表
db.create_tables([Person, Pet])

2、初始化数据

def init_data():
    person1 = Person.create(name='Tom', birthday='2020-01-01')
    Pet.create(owner=person1, name='Dog')
    Pet.create(owner=person1, name='Cat')
    person2 = Person.create(name='Jack', birthday='2020-01-02')
    Pet.create(owner=person2, name='Dog')
    Pet.create(owner=person2, name='Cat')
init_data()
"""
sqlite> select * from person;
id          name        birthday  
----------  ----------  ----------
1           Tom         2020-01-01
2           Jack        2020-01-02
sqlite> select * from pet;
id          owner_id    name        create_time          update_time               
----------  ----------  ----------  -------------------  -------------------
1           1           Dog         2021-03-02 10:16:07  2021-03-02 10:16:07
2           1           Cat         2021-03-02 10:16:07  2021-03-02 10:16:07
3           2           Dog         2021-03-02 10:36:01  2021-03-02 10:36:01
4           2           Cat         2021-03-02 10:36:01  2021-03-02 10:36:01

3、N+1问题

3-1、一对多,取列表

for row in Person.select():
   print(row.name)
    for pet in row.pets:
        print(pet.name)
"""
取N+1次 : 先取一次person列表;然后一个Person对象,取一次pets列表
('SELECT "t1"."id", "t1"."name", "t1"."birthday" FROM "person" AS "t1"', [])
('SELECT "t1"."id", "t1"."owner_id", "t1"."name", "t1"."create_time", "t1"."update_time" FROM "pet" AS "t1" WHERE ("t1"."owner_id" = ?)', [1])
('SELECT "t1"."id", "t1"."owner_id", "t1"."name", "t1"."create_time", "t1"."update_time" FROM "pet" AS "t1" WHERE ("t1"."owner_id" = ?)', [2])
"""
# 优化后方法
users = Person.select()
pets = Pet.select()
users_with_pets = prefetch(users, pets)
for row in users_with_pets:
    print(row.name)
    for pet in row.pets:
        print(pet.name)
"""
固定取两次:一次person列表;一次pet列表
('SELECT "t1"."id", "t1"."owner_id", "t1"."name", "t1"."create_time", "t1"."update_time" FROM "pet" AS "t1" WHERE ("t1"."owner_id" IN (SELECT "t2"."id" FROM "person" AS "t2"))', [])
('SELECT "t1"."id", "t1"."name", "t1"."birthday" FROM "person" AS "t1"', [])
"""

3-2、一对一查询,取列表

pets = Pet.select()
for pet in pets:
    print(pet.name, pet.owner.name)
"""
N+1次查询:首先取pet列表;逐个取pet对应的person
('SELECT "t1"."id", "t1"."owner_id", "t1"."name", "t1"."create_time", "t1"."update_time" FROM "pet" AS "t1"', [])
('SELECT "t1"."id", "t1"."name", "t1"."birthday" FROM "person" AS "t1" WHERE ("t1"."id" = ?) LIMIT ? OFFSET ?', [1, 1, 0])
('SELECT "t1"."id", "t1"."name", "t1"."birthday" FROM "person" AS "t1" WHERE ("t1"."id" = ?) LIMIT ? OFFSET ?', [1, 1, 0])
('SELECT "t1"."id", "t1"."name", "t1"."birthday" FROM "person" AS "t1" WHERE ("t1"."id" = ?) LIMIT ? OFFSET ?', [2, 1, 0])
('SELECT "t1"."id", "t1"."name", "t1"."birthday" FROM "person" AS "t1" WHERE ("t1"."id" = ?) LIMIT ? OFFSET ?', [2, 1, 0])
"""
# 优化方法
pets = Pet.select(Pet, Person).join(Person)
for pet in pets:
    print(pet.name, pet.owner.name)
"""
固定取1次
('SELECT 
"t1"."id", "t1"."owner_id", "t1"."name", "t1"."create_time", "t1"."update_time", 
"t2"."id", "t2"."name", "t2"."birthday" 
FROM "pet" AS "t1" INNER JOIN "person" AS "t2" 
ON ("t1"."owner_id" = "t2"."id")', [])
"""

打印SQL执行耗时

from functools import wraps
import time
import logging
from peewee import MySQLDatabase, SENTINEL, Model
logger = logging.getLogger('peewee')
# 计时器
def timer(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        ret = func(*args, **kwargs)
        end_time = time.time()
        logger.debug("time: %.2f s" % (end_time - start_time))
        return ret
    return wrapper
class CustomMySQLDatabase(MySQLDatabase):
    """
    打印sql执行时间
    see: https://github.com/coleifer/peewee/issues/2370
    """
    @timer
    def execute_sql(self, sql, params=None, commit=SENTINEL):
        return super().execute_sql(sql, params, commit)

支持原样查询返回dict字典对象

使用cator 模块

pip install cator

doc: https://github.com/mouday/cator

from peewee import MySQLDatabase
from cator import DatabaseProxy
db = MySQLDatabase(**config)
db_proxy = DatabaseProxy(db)

单个模型数据转dict字典

from copy import deepcopy
class BaseModel(Model):
    def to_dict(self):
        """model to dict"""
        return deepcopy(self.__data__)
    class Meta:
        database = db

文档:avoiding-the-n-1-problem

相关实践学习
每个IT人都想学的“Web应用上云经典架构”实战
本实验从Web应用上云这个最基本的、最普遍的需求出发,帮助IT从业者们通过“阿里云Web应用上云解决方案”,了解一个企业级Web应用上云的常见架构,了解如何构建一个高可用、可扩展的企业级应用架构。
MySQL数据库入门学习
本课程通过最流行的开源数据库MySQL带你了解数据库的世界。   相关的阿里云产品:云数据库RDS MySQL 版 阿里云关系型数据库RDS(Relational Database Service)是一种稳定可靠、可弹性伸缩的在线数据库服务,提供容灾、备份、恢复、迁移等方面的全套解决方案,彻底解决数据库运维的烦恼。 了解产品详情: https://www.aliyun.com/product/rds/mysql 
相关文章
机器学习/深度学习 算法 自动驾驶
134 0
|
25天前
|
存储 人工智能 算法
Python实现简易成语接龙小游戏:从零开始的趣味编程实践
本项目将中国传统文化与编程思维相结合,通过Python实现成语接龙游戏,涵盖数据结构、算法设计与简单AI逻辑,帮助学习者在趣味实践中掌握编程技能。
98 0
|
1月前
|
大数据 数据处理 数据安全/隐私保护
Python3 迭代器与生成器详解:从入门到实践
简介:本文深入解析Python中处理数据序列的利器——迭代器与生成器。通过通俗语言与实战案例,讲解其核心原理、自定义实现及大数据处理中的高效应用。
75 0
|
2月前
|
数据采集 Web App开发 JSON
Python爬虫基本原理与HTTP协议详解:从入门到实践
本文介绍了Python爬虫的核心知识,涵盖HTTP协议基础、请求与响应流程、常用库(如requests、BeautifulSoup)、反爬应对策略及实战案例(如爬取豆瓣电影Top250),帮助读者系统掌握数据采集技能。
192 0
|
2月前
|
传感器 数据采集 监控
Python生成器与迭代器:从内存优化到协程调度的深度实践
简介:本文深入解析Python迭代器与生成器的原理及应用,涵盖内存优化技巧、底层协议实现、生成器通信机制及异步编程场景。通过实例讲解如何高效处理大文件、构建数据流水线,并对比不同迭代方式的性能特点,助你编写低内存、高效率的Python代码。
119 0
|
2月前
|
人工智能 自然语言处理 安全
Python构建MCP服务器:从工具封装到AI集成的全流程实践
MCP协议为AI提供标准化工具调用接口,助力模型高效操作现实世界。
438 1
|
2月前
|
传感器 大数据 API
Python数字限制在指定范围内:方法与实践
在Python编程中,限制数字范围是常见需求,如游戏属性控制、金融计算和数据过滤等场景。本文介绍了五种主流方法:基础条件判断、数学运算、装饰器模式、类封装及NumPy数组处理,分别适用于不同复杂度和性能要求的场景。每种方法均有示例代码和适用情况说明,帮助开发者根据实际需求选择最优方案。
79 0
|
2月前
|
API 数据安全/隐私保护 开发者
Python自定义异常:从入门到实践的轻松指南
在Python开发中,自定义异常能提升错误处理的精准度与代码可维护性。本文通过银行系统、电商库存等实例,详解如何创建和使用自定义异常,涵盖异常基础、进阶技巧、最佳实践与真实场景应用,助你写出更专业、易调试的代码。
102 0
|
9月前
|
存储 人工智能 运维
【01】做一个精美的打飞机小游戏,浅尝阿里云通义灵码python小游戏开发AI编程-之飞机大战小游戏上手实践-优雅草央千澈-用ai开发小游戏尝试-分享源代码和游戏包
【01】做一个精美的打飞机小游戏,浅尝阿里云通义灵码python小游戏开发AI编程-之飞机大战小游戏上手实践-优雅草央千澈-用ai开发小游戏尝试-分享源代码和游戏包
477 48
【01】做一个精美的打飞机小游戏,浅尝阿里云通义灵码python小游戏开发AI编程-之飞机大战小游戏上手实践-优雅草央千澈-用ai开发小游戏尝试-分享源代码和游戏包
|
10月前
|
数据可视化 算法 数据挖掘
Python量化投资实践:基于蒙特卡洛模拟的投资组合风险建模与分析
蒙特卡洛模拟是一种利用重复随机抽样解决确定性问题的计算方法,广泛应用于金融领域的不确定性建模和风险评估。本文介绍如何使用Python和EODHD API获取历史交易数据,通过模拟生成未来价格路径,分析投资风险与收益,包括VaR和CVaR计算,以辅助投资者制定合理决策。
551 15

推荐镜像

更多