文檔: 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/catorfrom 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