インストール

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('inventory_db', host="localhost", port=3306, user="admin", password="secure_password")


class BaseModel(Model):
    created_at = DateTimeField(default=datetime.now, verbose_name="作成日時")
    class Meta:
        database = db

class Vendor(BaseModel):
    company_name = CharField(max_length=100, verbose_name="会社名", index=True)
    location = CharField(max_length=100, verbose_name="所在地")
    contact_number = CharField(max_length=11, verbose_name="連絡先")
    class Meta:
        database = db
        table_name = "vendor"

class Product(BaseModel):
    vendor = ForeignKeyField(Vendor, verbose_name="販売元", backref="products")
    product_name = CharField(max_length=100, verbose_name="製品名", index=True)
    view_count = IntegerField(default=0, verbose_name="閲覧数")
    stock_quantity = IntegerField(default=0, verbose_name="在庫数")
    unit_price = FloatField(default=0.0, verbose_name="単価")
    description = TextField(verbose_name="製品説明")

    class Meta:
        table_name = "product"


def initialize_database():
    db.create_tables([Product, Vendor])
    
if __name__ == "__main__":
    initialize_database()

このスクリプトを実行すると、データベースに2つのテーブルが作成されます。

追加するデータ data.py

vendor_data = [
    {
        "company_name":"楽天市場",
        "location":"東京都",
        "contact_number":"09011112222"
    },
    {
        "company_name":"ヤフーショッピング",
        "location":"大阪府",
        "contact_number":"09033334444"
    },
    {
        "company_name":"Amazon",
        "location":"神奈川県",
        "contact_number":"09055556666"
    }
]

product_data = [
    {
        "vendor":1,
        "product_name": "キリン 一番搾り 350ml缶 24本",
        "view_count": 250,
        "stock_quantity": 120,
        "unit_price": 4500,
        "description": "キリンビールの定番商品。新鮮な原料を使用した爽やかな味わいが特徴です。"
    },
    {
        "vendor":2,
        "product_name": "サントリー ザ・プレミアム・モルツ 350ml缶 24本",
        "view_count": 180,
        "stock_quantity": 80,
        "unit_price": 5800,
        "description": "サントリーの高級ビール。豊かな香りとまろやかな味わいが魅力です。"
    },
    {
        "vendor":3,
        "product_name": "アサヒ スーパードライ 350ml缶 24本",
        "view_count": 320,
        "stock_quantity": 150,
        "unit_price": 4200,
        "description": "アサヒビールの人気商品。クリーンな味わいと爽快感が特徴です。"
    },
    {
        "vendor":1,
        "product_name": "キリン 淡麗グリーンラベル 350ml缶 24本",
        "view_count": 95,
        "stock_quantity": 60,
        "unit_price": 3800,
        "description": "キリンビールのスタンダード商品。バランスの取れた味わいが魅力です。"
    },
    {
        "vendor":2,
        "product_name": "サントリー ウイスキー 角瓶 700ml",
        "view_count": 75,
        "stock_quantity": 30,
        "unit_price": 8500,
        "description": "サントリーのウイスキー。豊かな香りと深い味わいが特徴です。"
    }
]

コードを表示データベースへのデータ挿入 (peewee_test.py)

from models.model import Vendor, Product
from data import vendor_data, product_data

def save_data():
    for vendor_info in vendor_data:
        vendor = Vendor()
        vendor.company_name = vendor_info["company_name"]
        vendor.location = vendor_info["location"]
        vendor.contact_number = vendor_info["contact_number"]

        vendor.save()

    for product_info in product_data:
        product = Product(**product_info)
        product.save()

if __name__ == "__main__":
    save_data()

スクリプトを実行し、データベースを確認します。

データのクエリ (peewee_test.py)

from models.model import Vendor, Product
from data import vendor_data, product_data
def query_data():
    #単一データの取得
    #product = Product.get(Product.id==1)
    product = Product.get_by_id(1)
    product = Product[1]

    #selectはModelSelectオブジェクトを返す
    #すべてのデータを取得
    # select price from product
    products = Product.select(Product.product_name, Product.unit_price) # 遅延読み込み、イテレーション時にのみデータベースをクエリ

    # select * from product where price > 1000
    products = Product.select().where(Product.unit_price>1000)

    #select * from product where price>1000 and view_count>200
    products = Product.select().where((Product.unit_price>1000)|(Product.view_count>200))

    #select * from product where product_name like "%ラベル"
    products = Product.select().where(Product.product_name.contains("ラベル"))

    products = Product.select().where(Product.id<<[1,3])
    products = Product.select().where((Product.id==1)|(Product.id==3))
    products = Product.select().where((Product.id.in_([1,3])))


    #select * from product where price>view_count
    products = Product.select().where(Product.unit_price>Product.view_count)

    #ソート select * from product order by price desc
    products = Product.select().order_by(Product.unit_price.asc())
    products = Product.select().order_by(Product.unit_price)

    #ページネーション
    products = Product.select().order_by(Product.unit_price).paginate(2, 2)
    for product in products:
        print(product.unit_price)


if __name__ == "__main__":
    query_data()<br></br>

データの更新 (peewee_test.py)

更新時にはexecuteを呼び出して遅延読み込みを即時実行する必要があります。

from models.model import Vendor, Product
from data import vendor_data, product_data

from peewee import ModelUpdate
def update_data():
    try:
        # update view_count=100 where id =1
        Product.update(view_count=Product.view_count+1).where(Product.id==1).execute()

        product = Product.get_by_id(1)
        product.view_count += 1
        product.save()

    except Product.DoesNotExist:
        pass


if __name__ == "__main__":
    update_data()

データの削除 (peewee_test.py)

更新時と同様に、executeを呼び出して遅延読み込みを即時実行する必要があります。

from models.model import Vendor, Product
from data import vendor_data, product_data

from peewee import ModelUpdate
def delete_data():
        # 削除
        product = Product.get_by_id(1)
        product.delete_instance()
        # delete from product where price>5000
        Product.delete().where(Product.unit_price > 5000).execute()


if __name__ == "__main__":
    delete_data()

peewee-asyncをTornadoに統合する

peewee-asyncを統合する理由は、peeweeに多くの同期メソッド（例：更新、保存、削除時のexecute）が含まれているためです。これらを非同期メソッドにする必要があります。peewee-asyncはasyncioが提供するpeewee ORMの非同期インターフェースで、Tornadoで直接使用できます。

peewee-asyncのインストール

pip install --pre peewee-async　

GitHubリポジトリ

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(
    'inventory_db', host="localhost", port=3306, user="admin", password="secure_password"
)

objects = peewee_async.Manager(database)

# 同期は不要になりました！


database.set_allow_sync(False)


class BaseModel(Model):
    created_at = DateTimeField(default=datetime.now, verbose_name="作成日時")
    class Meta:
        database = database

class Vendor(BaseModel):
    company_name = CharField(max_length=100, verbose_name="会社名", index=True)
    location = CharField(max_length=100, verbose_name="所在地")
    contact_number = CharField(max_length=11, verbose_name="連絡先")
    class Meta:
        database = database
        table_name = "vendor"

class Product(BaseModel):
    vendor = ForeignKeyField(Vendor, verbose_name="販売元", backref="products")
    product_name = CharField(max_length=100, verbose_name="製品名", index=True)
    view_count = IntegerField(default=0, verbose_name="閲覧数")
    stock_quantity = IntegerField(default=0, verbose_name="在庫数")
    unit_price = FloatField(default=0.0, verbose_name="単価")
    description = TextField(verbose_name="製品説明")

    class Meta:
        table_name = "product"


def initialize_database():
    db.create_tables([Product, Vendor])
if __name__ == "__main__":
    initialize_database()

peewee_async_test.py

import asyncio

from models.model import Product
from models.model import objects


async def async_handler():
    await objects.create(Product, vendor_id=1, product_name="サントリー ウイスキー 角瓶 700ml",
                         view_count=15, stock_quantity=50, unit_price=8500, description="サントリーのウイスキー。豊かな香りと深い味わいが特徴です。")
    products = await objects.execute(Product.select())
    for product in products:
        print(product.product_name)

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

実行結果は以下の通りです。