Sanic作为一个支持 async/await 语法的高性能，异步无阻塞框架，当我们的框架中需要使用MySQL时，不建议安装pymysql进行数据库操作，而是优先去选择异步无阻塞的第三方库，例如：aiomysql。

aiomysql的安装：pip install aiomysql
sqlalchemy的安装 pip install sqlalchemy创建表结构：

CREATE TABLE userinfo (
id INT AUTO_INCREMENT PRIMARY KEY,
create_time datetime NOT NULL,
update_time datetime NOT NULL,
username VARCHAR ( 20 ) NOT NULL,
password VARCHAR ( 32 ) NOT NULL,
user_type INT NOT NULL,
mobile VARCHAR ( 11 ) NOT NULL,
real_name VARCHAR ( 20 ) NOT NULL
);

model.py

import sqlalchemy as sa

metadata = sa.MetaData()

userinfo = sa.Table(
    'userinfo',
    metadata,
    sa.Column('id', sa.Integer, autoincrement=True, primary_key=True),
    sa.Column('create_time', sa.DATETIME(6), nullable=True),
    sa.Column('update_time', sa.DATETIME(6), nullable=True),

    sa.Column('username', sa.String(20), nullable=False),
    sa.Column('password', sa.String(32), nullable=True),
    sa.Column('user_type', sa.Integer, nullable=False),
    sa.Column('mobile', sa.String(11), nullable=False),
    sa.Column('real_name', sa.String(20), nullable=True)
)

config/dbconfig.py

DBINFO = {
    'NAME': 'testdb',
    'USER': 'root',
    'PASSWORD': '123456',
    'HOST': '127.0.0.1',
    'PORT': '3306'
}

demo.py

import asyncio
import datetime
from aiomysql.sa import create_engine
from config.dbconfig import DBINFO
from model import userinfo


async def main():
    engine = await create_engine(user=DBINFO["USER"], db=DBINFO["NAME"],
                                 host=DBINFO["HOST"], password=DBINFO["PASSWORD"])
    async with engine.acquire() as conn:
        create_time = str(datetime.datetime.now())[:-7]
        update_time = str(datetime.datetime.now())[:-7]

        username = "Tom"
        password = "abc123456"
        user_type = 1
        mobile = "18866668888"
        real_name = "汤姆"

        await conn.execute(userinfo.insert().values(create_time=create_time, update_time=update_time, username=username,
                                                    password=password, user_type=user_type, mobile=mobile,
                                                    real_name=real_name
                                                    ))
        await conn.execute('commit')
        
        # 注意这里需要使用 await 接收，不然会抛出异常
        r2 = await conn.execute(userinfo.select())
        async for row in r2:
            print(row.create_time, row.username)

    engine.close()
    await engine.wait_closed()


if __name__ == '__main__':
    result = main()
    asyncio.run(result)

最终运行结果：

图片