04 | 在tornado中 使用peewee作為orm

安裝

pip install peewee　　

github 地址

https://github.com/coleifer/peewee　

官方文檔（文檔寫的很詳細，建議讀一遍）

http://docs.peewee-orm.com/en/latest/

　　

使用 peewee創建模型類

 實例化一個 數據庫的實例 db ，要在Meta 中指定

 models/model.py

from datetime import datetime

from peewee import *
from peewee import Model

db = MySQLDatabase('message', host="127.0.0.1", port=3306, user="root", password="mysql")


class BaseModel(Model):
    add_time = DateTimeField(default=datetime.now, verbose_name="添加時間")
    class Meta:
        database = db

class Supplier(BaseModel):
    name = CharField(max_length=100, verbose_name="名稱", index=True)
    address = CharField(max_length=100, verbose_name="聯系地址")
    phone = CharField(max_length=11, verbose_name="聯系方式")
    class Meta:
        database = db
        table_name = "supplier"

class Goods(BaseModel):
    supplier = ForeignKeyField(Supplier, verbose_name="商家", backref="goods")
    name = CharField(max_length=100, verbose_name="商品名稱", index=True)
    click_num = IntegerField(default=0, verbose_name="點擊數")
    goods_num = IntegerField(default=0, verbose_name="庫存數")
    price = FloatField(default=0.0, verbose_name="價格")
    brief = TextField(verbose_name="商品簡介")

    class Meta:
        table_name = "goods"


def init_table():
    db.create_tables([Goods, Supplier])
    
if __name__ == "__main__":
    init_table()

　

 運行該腳本，數據庫中多了兩張表

 

 要添加的數據data.py

supplier_list = [
    {
        "name":"淘寶",
        "address":"杭州市",
        "phone":"18888888888"
    },
    {
        "name":"京東",
        "address":"上海市",
        "phone":"17777777777"
    },
    {
        "name":"天貓",
        "address":"北京市",
        "phone":"16666666666"
    }
]

goods_list = [
    {
        "supplier":1,
        "name": "52度茅台集團國隆雙喜酒500mlx6",
        "click_num": 100,
        "goods_num": 666,
        "price": 128,
        "brief": "貴州茅台酒廠（集團）保健酒業有限公司生產，是以“龍”字打頭的酒水。中國龍文化上下8000年，源遠而流長，龍的形象是一種符號、一種意緒、一種血肉相聯的情感。"
    },
    {
        "supplier":2,
        "name": "52度水井坊臻釀八號500ml",
        "click_num": 585,
        "goods_num": 288,
        "price": 36,
        "brief": "貴州茅台酒廠（集團）保健酒業有限公司生產，是以“龍”字打頭的酒水。中國龍文化上下8000年，源遠而流長，龍的形象是一種符號、一種意緒、一種血肉相聯的情感。"
    },
    {
        "supplier":3,
        "name": "53度茅台仁酒500ml",
        "click_num": 553,
        "goods_num": 280,
        "price": 190,
        "brief": "貴州茅台酒廠（集團）保健酒業有限公司生產，是以“龍”字打頭的酒水。中國龍文化上下8000年，源遠而流長，龍的形象是一種符號、一種意緒、一種血肉相聯的情感。"
    },
    {
        "supplier":1,
        "name": "茅台53度飛天茅台500ml",
        "click_num": 48,
        "goods_num": 20,
        "price": 22,
        "brief": "貴州茅台酒廠（集團）保健酒業有限公司生產，是以“龍”字打頭的酒水。中國龍文化上下8000年，源遠而流長，龍的形象是一種符號、一種意緒、一種血肉相聯的情感。"
    },
    {
        "supplier":2,
        "name": "芝華士12年蘇格蘭威士忌700ml",
        "click_num": 31,
        "goods_num": 15,
        "price": 88,
        "brief": "貴州茅台酒廠（集團）保健酒業有限公司生產，是以“龍”字打頭的酒水。中國龍文化上下8000年，源遠而流長，龍的形象是一種符號、一種意緒、一種血肉相聯的情感。"
    }
]

 

peewee_test.py  向數據庫中插入數據 

from models.model import Supplier, Goods
from data import supplier_list, goods_list

def save_model():
    for data in supplier_list:
        supplier = Supplier()
        supplier.name = data["name"]
        supplier.address = data["address"]
        supplier.phone = data["phone"]

        supplier.save()

    for data in goods_list:
        good = Goods(**data)
        good.save()

if __name__ == "__main__":
    save_model()

 運行腳本，查看數據庫

 

 

 peewee_test.py  查詢數據

from models.model import Supplier, Goods
from data import supplier_list, goods_list
def query_model():
    #獲取某一條數據
    #good = Goods.get(Goods.id==1)
    good = Goods.get_by_id(1)
    good = Goods[1]

    #select 返回的是modelselect對象
    #獲取所有數據
    # select price from goods
    goods = Goods.select(Goods.name, Goods.price) # 懶加載，只有在迭代的時候才會查詢數據庫

    # select * from goods where price > 100
    goods = Goods.select().where(Goods.price>100)

    #select * from goods where price>100 and click_num>200
    goods = Goods.select().where((Goods.price>100)|(Goods.click_num>200))

    #select * from goods where name like "%飛天"
    goods = Goods.select().where(Goods.name.contains("飛天"))

    goods = Goods.select().where(Goods.id<<[1,3])
    goods = Goods.select().where((Goods.id==1)|(Goods.id==3))
    goods = Goods.select().where((Goods.id.in_([1,3])))


    #select * from goods where price>click_num
    goods = Goods.select().where(Goods.price>Goods.click_num)

    #排序 select * from goods order by price desc
    goods = Goods.select().order_by(Goods.price.asc())
    goods = Goods.select().order_by(Goods.price)

    #分頁
    goods = Goods.select().order_by(Goods.price).paginate(2, 2)
    for good in goods:
        print(good.price)


if __name__ == "__main__":
    query_model()

 

 peewee_test.py  更新

在更新的時候要執行execute 懶加載，讓它立即執行

from models.model import Supplier, Goods
from data import supplier_list, goods_list

from peewee import ModelUpdate
def update_model():
    try:
        # update click_num=100 where id =1
        Goods.update(click_num=Goods.click_num+1).where(Goods.id==1).execute()

        good = Goods.get_by_id(1)
        good.click_num += 1
        good.save()

    except Goods.DoesNotExist:
        pass


if __name__ == "__main__":
    update_model()

 

peewee_test.py  刪除

在更新的時候要執行execute 懶加載，讓它立即執行

from models.model import Supplier, Goods
from data import supplier_list, goods_list

from peewee import ModelUpdate
def update_model():
        # 刪除
        good = Goods.get_by_id(1)
        good.delete_instance()
        # delete from goods where price>150
        Goods.delete().where(Goods.price > 150).execute()


if __name__ == "__main__":
    update_model()

 

通過peewee-async集成到tornado中

之所以集成peewee-async是因為peewee中存在大量的同步方法 execute （比如在更新和保存，刪除），而我們需要異步的方法，peewee-async 中由asyncio提供的peewee ORM的異步接口，可以直接在tornado中使用 

安裝 peewee-async　

pip install --pre peewee-async　

 gihub 地址

https://github.com/05bit/peewee-async

　　

根據官方文檔改寫models/model.py

from datetime import datetime

from peewee import *
from peewee import Model
import peewee_async

database = peewee_async.MySQLDatabase(
    'message', host="127.0.0.1", port=3306, user="root", password="mysql"
)

objects = peewee_async.Manager(database)

# No need for sync anymore!


database.set_allow_sync(False)


class BaseModel(Model):
    add_time = DateTimeField(default=datetime.now, verbose_name="添加時間")
    class Meta:
        database = database

class Supplier(BaseModel):
    name = CharField(max_length=100, verbose_name="名稱", index=True)
    address = CharField(max_length=100, verbose_name="聯系地址")
    phone = CharField(max_length=11, verbose_name="聯系方式")
    class Meta:
        database = database
        table_name = "supplier"

class Goods(BaseModel):
    supplier = ForeignKeyField(Supplier, verbose_name="商家", backref="goods")
    name = CharField(max_length=100, verbose_name="商品名稱", index=True)
    click_num = IntegerField(default=0, verbose_name="點擊數")
    goods_num = IntegerField(default=0, verbose_name="庫存數")
    price = FloatField(default=0.0, verbose_name="價格")
    brief = TextField(verbose_name="商品簡介")

    class Meta:
        table_name = "goods"


def init_table():
    db.create_tables([Goods, Supplier])
if __name__ == "__main__":
    init_table()

 

peewee_async_test.py

import asyncio

from chapter04.models.model import Goods
from chapter04.models.model import objects


async def handler():
    await objects.create(Goods, supplier_id=1, name="53度水井坊臻釀八號500ml",
                         click_num=20, goods_num=1000, price=500, brief="州茅台酒廠（集團）保健酒業有限公司生產")
    goods = await objects.execute(Goods.select())
    for good in goods:
        print(good.name)

loop = asyncio.get_event_loop()
loop.run_until_complete(handler())

輸出結果如下