torpeewee -- fixed example from torpeewee/README.rst -- async peewee for Tornado

torpeewee_ex01.py

# -*- coding: utf-8 -*-
# REWORKED by biggers@utsl.com, from: 
# 16/7/7
# create by: snower

import datetime
from tornado import gen
from tornado.ioloop import IOLoop
from torpeewee import (
    Model,
    MySQLDatabase,
    IntegerField,
    CharField,
    DateTimeField,
)

_CREAT = True
mysql_db = MySQLDatabase(None)


class BaseModel(Model):
    class Meta:
        database = mysql_db


class TorpwTest(BaseModel):
    data = CharField(max_length=64, null=False)
    count = IntegerField(default=0)
    created_at = DateTimeField()
    updated_at = DateTimeField()


@mysql_db.transaction()
@gen.coroutine
def run_transaction(transaction):
    for i in (yield TorpwTest.use(transaction).
              select().order_by(TorpwTest.id.desc()).limit(2)):
        print(i.id, i.data)

    print()
    t = yield TorpwTest.use(transaction).\
        create(data="test",
               created_at=datetime.datetime.now(),
               updated_at=datetime.datetime.now())
    print(t.id, t.data)

    for i in (yield TorpwTest.select().order_by(TorpwTest.id.desc()).limit(2)):
        print(i.id, i.data)

    print()
    for i in (yield TorpwTest.use(transaction).
              select().order_by(TorpwTest.id.desc()).limit(2)):
        print(i.id, i.data)


@gen.coroutine
def run():
    """ """
    global _CREAT

    mysql_db.init(database="adminstack",
                  host="127.0.0.1", port=3306,
                  user="mrsql", passwd="mysqueel",
                  charset='utf8')

    mysql_db.connect()

    if _CREAT:
        yield TorpwTest.drop_table()
        yield TorpwTest.create_table()

    t = yield TorpwTest.select().where(TorpwTest.id == 5).first()
    print(t)

    c = yield TorpwTest.select().where(TorpwTest.id > 5).count()
    print(c)

    c = yield TorpwTest.select().\
        where(TorpwTest.id > 5).group_by(TorpwTest.data).count()
    print(c)

    for i in (yield TorpwTest.select().
              where(TorpwTest.id > 5).where(TorpwTest.id <= 10)):
        print(i.id, i.data)

    for i in (yield TorpwTest.select().order_by(TorpwTest.id.desc()).limit(2)):
        print(i.id, i.data)
    t = yield TorpwTest.create(data="test",
                               created_at=datetime.datetime.now(),
                               updated_at=datetime.datetime.now())
    print(t.id, t.data)
    for i in (yield TorpwTest.select().order_by(TorpwTest.id.desc()).limit(2)):
        print(i.id, i.data)

    print()

    # TypeError: 'ModelSelectFuture' object does not support indexing
    t = yield TorpwTest.select().order_by(TorpwTest.id.desc()).limit(1)
    print(t[0].id, t[0].data, t[0].count)
    t.count += 1

    t = yield TorpwTest.select().order_by(TorpwTest.id.desc()).limit(1)
    print(t[0].id, t[0].data, t[0].count)

    print()

    with (yield mysql_db.transaction()) as transx:
        for i in (yield TorpwTest.use(transx).
                  select().order_by(TorpwTest.id.desc()).limit(2)):
            print(i.id, i.data)

        print()
        t = yield TorpwTest.use(transx).\
            create(data="test",
                   created_at=datetime.datetime.now(),
                   updated_at=datetime.datetime.now())
        print(t.id, t.data)

        for i in (yield TorpwTest.select().
                  order_by(TorpwTest.id.desc()).limit(2)):
            print(i.id, i.data)

        print()
        for i in (yield TorpwTest.use(transx).
                  select().order_by(TorpwTest.id.desc()).limit(2)):
            print(i.id, i.data)

    print()

    yield run_transaction()


ioloop = IOLoop.instance()
ioloop.run_sync(run)