天天看點

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