sanic学习--初步使用

上次介绍sanic还是在两年前，当时的sanic还处于高速迭代阶段，然而不知不觉，sanic在这两年中更新了不少内容。

本次博客中使用的sanic版本是22.3.2版本，所以很多sanic的用法已经发生了变化。sanic框架的文档见这里

项目结构：

- api    # 项目接口文件夹
- config # 项目配置文件夹
- model  # 项目模型文件夹，对应着数据库模型
- utils  # 工具函数
- requestment.txt # 项目包依赖
- server.py # 项目启动文件

首先看一下目前的项目包依赖：

//requrestment.txt
aiofiles==0.8.0
aiomysql==0.1.1
httptools==0.4.0
multidict==6.0.2
peewee==3.15.0
peewee-async==0.7.2
PyMySQL==1.0.2
sanic==22.3.2
sanic-routing==22.3.0
ujson==5.3.0
uvloop==0.16.0
websockets==10.3

项目使用到了sanic + peewee + peewee-async来进行数据的处理，数据库使用的是mysql，因此也安装了mysql相应的python库，如aiomysql和PyMySQL

然后看一下目前项目文件

# server.py
from sanic import Sanic
from api.user import user_bp
from peewee_async import Manager, PooledMySQLDatabase
from model import db
from config import *

# 创建app实例
app = Sanic("MyApp")
# 注册user蓝图
app.blueprint(user_bp)

# 注册服务启动之前的监听器，用于服务器前的的相关配置，如初始化数据库连接，redis连接等
@app.before_server_start
async def setup(app, loop):
    pool = PooledMySQLDatabase(db_config.pop('database'), **db_config)
    db.initialize(pool)
    # 将数据库manager注册到app的ctx上下文中
    app.ctx.db = Manager(db)

# 注册服务结束后监听器，关闭相应的连接，如数据库，redis的连接等
@app.after_server_stop
async def stop(app):
  	# 关闭数据库连接
    await app.ctx.db.close()


if __name__ == '__main__':
    app.run()

该项目文件主要做的任务就是创建app实例，以及初始化实例相关的蓝图和项目中用到的数据库，redis等

需要注意的是，这里只是简单的一个例子，通常我们会创建多个blueprint，然后注册到app中，也会使用中间件来进行数据权限或者针对服务异常的控制。

在该服务中使用到了user的一个蓝图，user_bp

那么让我们来看一下api的定义：

# api/user.py
from sanic.response import json
from sanic import Blueprint

from model.auth import User
# 定义路由并设置路由前缀
user_bp = Blueprint('user', url_prefix='/user')

# 定义用户列表接口，通过使用bp的get定义方法类型
@user_bp.get('/list')
async def user_list(req):
  	# 获取到数据库的连接
    db = req.app.ctx.db
    # 获取用户列表
    users = await db.execute(User.select().dicts())
    # 通过json方法，返回接口内容
    return json(dict(code=0, msg="获取用户列表成功", data=[user for user in users]))

这个时候启动服务，然后调用http://127.0.0.1:8000/user/list，就可以看到服务调用了该接口。当然要保证该接口正常运行，需要创建User模型，以及创建mysql中对应的表结构。

此时可以看一下User模型的定义

# model/auth.py
default_time = SQL('DEFAULT CURRENT_TIMESTAMP')
update_time = SQL('ON UPDATE CURRENT_TIMESTAMP')
# 使用db代理，这样可以重复使用db
db = Proxy()

class User(Model):
    class Meta:
        table_name = "db_user"
        database = db
        indexes = (
            (('user_name', 'user_role'), True)
        )

    user_id = AutoField(help_text="用户id")
    user_code = CharField(max_length=16, help_text="用户编码")
    user_name = CharField(max_length=16, help_text="用户名称")
    user_pwd = CharField(null=True, help_text="用户密码")
    user_phone = CharField(null=True, help_text="用户手机号")
    user_role = ForeignKeyField(UserRole, backref="user_role", help_text="用户角色类型")
    create_time = DateTimeField(constraints=[default_time], help_text="角色创建时间")
    update_time = DateTimeField(null=True, constraints=[update_time], help_text="用户更新时间")

sanic的基本使用就到这里了， 后续就可以愉快的开发各种接口了