import asyncio
import peewee
import peewee_async
database = peewee_async.PostgresqlDatabase('test')
loop = asyncio.get_event_loop()
class TestModel(peewee_async.AioModel):
text = peewee.CharField()
class Meta:
database = database
# Create table synchronously!
TestModel.create_table(True)
# This is optional: close sync connection
database.close()
async def my_handler():
obj1 = TestModel.create(text="Yo, I can do it sync!")
obj2 = await TestModel.aio_create(text="Not bad. Watch this, I'm async!")
all_objects = await TestModel.select().aio_execute()
for obj in all_objects:
print(obj.text)
await TestModel.delete().aio_execute()
loop.run_until_complete(database.connect_async(loop=loop))
loop.run_until_complete(my_handler())