sanic/examples/sanic_aiopeewee.py

from sanic import Sanic
from sanic.response import json

from aiopeewee import AioModel, AioMySQLDatabase, model_to_dict
from peewee import CharField, TextField, DateTimeField
from peewee import ForeignKeyField, PrimaryKeyField


db = AioMySQLDatabase('test', user='root', password='',
                      host='127.0.0.1', port=3306)


class User(AioModel):
    username = CharField()

    class Meta:
        database = db


class Blog(AioModel):
    user = ForeignKeyField(User)
    title = CharField(max_length=25)
    content = TextField(default='')
    pub_date = DateTimeField(null=True)
    pk = PrimaryKeyField()

    class Meta:
        database = db


app = Sanic(__name__)


@app.listener('before_server_start')
async def setup(app, loop):
    # create connection pool
    await db.connect(loop)
    # create table if not exists
    await db.create_tables([User, Blog], safe=True)


@app.listener('before_server_stop')
async def stop(app, loop):
    # close connection pool
    await db.close()


@app.post('/users')
async def add_user(request):
    user = await User.create(**request.json)
    return json(await model_to_dict(user))


@app.get('/users/count')
async def user_count(request):
    count = await User.select().count()
    return json({'count': count})


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=8000)