Skip to content

Instantly share code, notes, and snippets.

@biggers biggers/torpeewee_ex01.py
Last active May 18, 2018

Embed
What would you like to do?
torpeewee -- fixed example from torpeewee/README.rst -- async peewee for Tornado
# -*- 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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.