sanic 教程

news2025/7/11 12:33:29

sanic 教程

在Sanic的生命周期流程大致如下:
http请求——Sanic解析request——匹配路由——请求中间件——视图函数——响应中间件——http响应

依赖库

sanic==21.3.4
sanic-jwt==1.7.0
sanic-openapi==21.12.0
gunicorn==20.1.0
PyMySQL==1.0.2
aiomysql==0.1.1
DBUtils==1.3

依赖框架使用

sanic-openapi使用说明

1、设置文档title:API_TITLE

2、设置版本:API_VERSION

3、配置系统描述:API_DESCRIPTION
在这里插入图片描述

4、配置服务条款:API_TERMS_OF_SERVICE


5、配置邮件联系方式:API_CONTACT_EMAIL

6、配置许可证:API_LICENSE_NAME

7、配置许可证地址:API_LICENSE_URL

8、配置BASE_URL:API_HOST

9、配置接口统一的前缀:API_BASEPATH


也可单独配置

10、配置协议类型:API_SCHEMES


11、配置API必须通过身份验证进行访问:API_SECURITY、API_SECURITY_DEFINITIONS

12、配置API Key - in header

13、配置API Key - in query

14、配置OAuth2校验


15、配置是否展示接口末尾"/":API_URI_FILTER,默认不展示,slash展示,All则都展示


sanic-jwt使用说明

sanic-jwt 则是针对Sanic 开发的一个基于 PyJWT 封装的 JWT 授权认证模块。

Configuration 参数



代码

from sanic import Sanic, request, response
from sanic_jwt import initialize, Configuration, Responses, protected, exceptions, Authentication, inject_user


class User:

    def __init__(self, uid, username, sex, password, info, black_level=0):
        self.user_id = uid
        self.sex = sex
        self.username = username
        self.password = password
        self.personal_info = info  # 只能登录后个人可见的信息
        self.black_level = black_level  # 黑名单等级,默认0为正常用户

    def __repr__(self):
        return "User(id='{}')".format(self.user_id)

    def to_dict(self):

        return {
            "uid": self.user_id,  # 注意:此处 "uid" 要与 MyJWTConfig 中的 user_id 设置一致!
            "sex": self.sex,
            "username": self.username,
            "personal_info": self.personal_info
        }


# 模拟一个用户列表
users = [
    User(1, "user1", "男", "123",  "这是仅 user1 可见信息", 1),
    User(2, "user2", "女", "456",  "这是仅 user2 可见信息", 0)
]

username_table = {u.username: u for u in users}
userid_table = {u.user_id: u for u in users}


async def authenticate(req: request.Request):
    username = req.json.get("username", None)
    password = req.json.get("password", None)

    if not username or not password:
        raise exceptions.AuthenticationFailed("用户名或密码为空!")
    user = username_table.get(username, None)
    if user is None:
        raise exceptions.AuthenticationFailed("用户名或密码不正确!")
    if password != user.password:
        raise exceptions.AuthenticationFailed("用户名或密码不正确!")
    return user


class MyJWTConfig(Configuration):
    # -------------- url_prefix ---------------------
    # [描述] 获取授权的路由地址
    # [默认] '/auth'
    url_prefix = '/login'

    # -------------- secret -------------------------
    # [描述] 加密密码
    # [默认] 'This is a big secret. Shhhhh'
    # [建议] 该密码是 JWT 的安全核心所在,需要保密,尽量使用更长更复杂的密码
    # |secret = ',xxx'

    # -------------- expiration_delta ----------------------
    # [描述] 过期时间,单位为秒
    # [默认] 30 分钟,即:60 * 30
    # [建议] 该时间不宜过长,同时建议开启 refresh_token_enabled 以便自动更新 token
    expiration_delta = 60 * 60  # 改为 10 分钟过期

    # -------------- cookie_set ---------------------
    # [描述] 是否将获取到的 token 信息写入到 cookie
    # [默认] False,即不写入cookie
    # 只有该项为 True,其它 cookie 相关设置才会起效。
    # cookie_set = True

    # -------------- cookie_access_token_name ---------------
    # [描述] cookie 中存储 token 的名称。
    # [默认] 'access_token'
    # cookie_access_token_name = "token"

    #  -------------- cookie_access_token_name ---------------
    # [描述] 包含用户 id 的用户对象的键或属性,这里对应 User 类的用户唯一标识
    # [默认] 'user_id'
    user_id = "uid"

    claim_iat = True  # 显示签发时间,JWT的默认保留字段,在 sanic-jwt 中默认不显示该项


class MyJWTAuthentication(Authentication):

    # 从 payload 中解析用户信息,然后返回查找到的用户
    # args[0]: request
    # args[1]: payload
    async def retrieve_user(self, *args, **kwargs):
        user_id_attribute = self.config.user_id()
        if not args or len(args) < 2 or user_id_attribute not in args[1]:
            return {}
        user_id = dict(args[1]).get(user_id_attribute)
        # TODO: 根据项目实际情况进行修改
        user = userid_table.get(user_id)
        return user

    # 拓展 payload
    async def extend_payload(self, payload, *args, **kwargs):
        # 可以获取 User 中的一些属性添加到 payload 中
        # 注意:payload 信息是公开的,这里不要添加敏感信息
        user_id_attribute = self.config.user_id()
        user_id = payload.get(user_id_attribute)
        # TODO: 根据项目实际情况进行修改
        user: User = userid_table.get(user_id)
        payload.update({'sex': user.sex})  # 比如添加性别属性
        return payload

    async def extract_payload(self, req, verify=True, *args, **kwargs):
        return await super().extract_payload(req, verify)


class MyJWTResponse(Responses):

    # 自定义发生异常的返回数据
    @staticmethod
    def exception_response(req: request.Request, exception: exceptions):
        # sanic-jwt.exceptions 下面定义的异常类型:
        # AuthenticationFailed
        # MissingAuthorizationHeader
        # MissingAuthorizationCookie
        # InvalidAuthorizationHeader
        # MissingRegisteredClaim
        # Unauthorized
        msg = str(exception)
        if exception.status_code == 500:
            msg = str(exception)
        elif isinstance(exception, exceptions.AuthenticationFailed):
            msg = str(exception)
        else:
            if "expired" in msg:
                msg = "授权已失效,请重新登录!"
            else:
                msg = "未授权,请先登录!"
        result = {
            "status": exception.status_code,
            "data": None,
            "msg": msg
        }
        return response.json(result, status=exception.status_code)


app = Sanic("my_auth_app")
initialize(app, authenticate=authenticate,
           authentication_class=MyJWTAuthentication, configuration_class=MyJWTConfig, responses_class=MyJWTResponse)


@app.route("/index")
@protected()  # 保护该路由,只有授权用户才能访问
async def protected_route_index(req: request.Request):
    # 从 request 中获取 payload,然后返回给前端
    payload = await req.app.auth.extract_payload(req)
    return response.json({'payloadInfo': payload})


@app.route("/info")
@inject_user()  # 注入用户信息
@protected()    # 保护该路由,只有授权用户才能访问
async def protected_route_info(req: request.Request, user: User):
    if user.black_level == 0:
        return response.json({'userName': user.username, "personalInfo": user.personal_info})
    else:  # 进入黑名单等级之后限制查看
        return response.json({'userName': user.username, "personalInfo": ""})


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080, auto_reload=True)

参考

https://swagger.io/docs/open-source-tools/swagger-ui/usage/configuration/

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/8781.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

WPS(WSC)中M1 到M8 图解

背景 之前实习的时候就学了Wifi p2p相关的东西&#xff0c;当时找M1到M8的功能把我累惨了&#xff0c;找到的还全是千篇一律的东西&#xff0c;讲的不是很清楚&#xff08;当然原版出书的那个前辈肯定是懂的&#xff09;&#xff0c;但是对我等小白不友好&#xff0c;就萌生了这…

3.线性代数-矩阵

矩阵和Tensor1. Tensor2.矩阵3.线性代数正确打开方式3.1 行视图3.2 列视图4.线性相关和线性无关5. Span、基和子空间(Subspace)6.四个基本的子空间6.1 列空间6.2 零空间6.3 行空间6.4 左零空间6.5 四个基本子空间的关系7.可逆矩阵8.方阵的特征值与特征向量9.特征分解9.1一般矩阵…

【Pytorch with fastai】第 7 章 :训练SOTA的模型

&#x1f50e;大家好&#xff0c;我是Sonhhxg_柒&#xff0c;希望你看完之后&#xff0c;能对你有所帮助&#xff0c;不足请指正&#xff01;共同学习交流&#x1f50e; &#x1f4dd;个人主页&#xff0d;Sonhhxg_柒的博客_CSDN博客 &#x1f4c3; &#x1f381;欢迎各位→点赞…

Nginx优化方案

目录 一、Nginx返回错误页面 1、HTTP常见状态代码列表 二、Nginx状态页面 1、安装status模块 2、激活status 三、优化并发连接数 1、压力测试软件ab&#xff08;http-tools&#xff09; 2、优化并发连接数 2.1、修改nginx并发数 2.2、修改内核最大文件数量 四、Nginx…

吉莱斯皮随机模拟算法(SSA)(Matlab代码实现)

&#x1f468;‍&#x1f393;个人主页&#xff1a;研学社的博客 &#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️&#x1f4a5;&#x1f4a5; &#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜…

【K8S系列】第十讲:Knative 简介

目录 一、 Serverless介绍 二、Knative 介绍 2.1 Knative 的定位 2.2 Knative的组成 2.2.1 Build 构建系统 2.2.2 Serving&#xff1a;服务系统 2.2.3 Eventing&#xff1a;事件系统 补充&#xff1a; 三、总结&#xff1a; 一、 Serverless介绍 在讲Knative之前&a…

【毕业设计】机器视觉手势检测和识别系统 - python 深度学习

文章目录0 前言1 实现效果2 技术原理2.1 手部检测2.1.1 基于肤色空间的手势检测方法2.1.2 基于运动的手势检测方法2.1.3 基于边缘的手势检测方法2.1.4 基于模板的手势检测方法2.1.5 基于机器学习的手势检测方法3 手部识别3.1 SSD网络3.2 数据集3.3 最终改进的网络结构4 最后0 前…

线程池源码解析 2.工作原理与内部结构

线程池源码解析—工作原理与内部结构 工作原理 概述 线程池是线程的池子&#xff0c;本质上是通过单个线程执行多个并发任务&#xff0c;使得尽量少的创建线程&#xff0c;减少开销。在线程池内部&#xff0c;是没有区分核心线程和非核心线程的&#xff0c;是通过 Set 集合的…

拒绝内卷,阿里架构师整理的这份Java核心手册,堪称最强

2022年注定是不寻常的一年&#xff0c;在今年因为疫情以及各大大厂纷纷传来裁员的消息&#xff0c;引得整个互联网圈动荡不堪。腾讯裁员30%、京东、百度、字节等大厂都在纷纷裁员&#xff0c;引的这些中厂和小厂也跟风裁员。 这个时候外部的各种变化愈发证明一个重要的一点&am…

2022.11.7-11.13 AI行业周刊(第123期):技术人员的职业发展在哪里?

篇章一&#xff1a;技术人员的职业发展 上周和大学时的舍友聊天&#xff0c;交流当前大家的生活状态。 我们已经本科毕业将近10年了&#xff0c;他目前也有两个孩子&#xff0c;在湖北的一个地级市中&#xff0c;从事的是通信行业。 不过随着工作的时间越久&#xff0c;他发…

软件测试面试真题 | 黑盒测试和白盒测试的基本概念是什么?

在软件测试的面试中&#xff0c;什么是黑盒测试&#xff0c;什么是白盒测试是特别容易被问到的一个问题。 面试官问出这个问题&#xff0c;其实考察的是大家对于软件测试基础理论的掌握程度。下面来梳理一下这个问题的回答思路。 黑盒测试 黑盒测试会把被测的软件看作是一个…

只会加班的项目经理,迟早被淘汰

早上好&#xff0c;我是老原。 最近看到一个文章的标题「废掉一个人最好的方式&#xff0c;就是让他忙到没时间学习」&#xff0c;具体内容是什么我还没有细读&#xff0c;只看完标题&#xff0c;有一丝心酸和自豪&#xff1a; 有那么一群人&#xff0c;在玻璃渣里找糖吃&…

皮带跑偏检测系统

皮带跑偏检测系统对皮带运行状态进行全天候实时监测&#xff0c;一旦皮带跑偏检测系统监测到现场皮带跑偏、撕裂、堆煤、异物等异常情况时&#xff0c;系统马上开展警报&#xff0c;通知后台&#xff0c;并提醒相关人员及时处置。皮带跑偏检测系统并把警报截屏和视频储存到数据…

附参考文献丨艾美捷Cholesterol胆固醇说明书

Cholesterol胆固醇以固体形式提供。可以通过将胆固醇溶解在所选择的溶剂中来制备储备溶液&#xff0c;该溶剂应使用惰性气体吹扫。胆固醇以约30mg/ml的浓度溶于有机溶剂氯-仿中。 艾美捷Cholesterol胆固醇参数&#xff1a; CAS号&#xff1a;57-88-5 正式名称&#xff1a;&am…

自动驾驶入门:预测

目录 概念 预测方式 障碍物预测 递归神经网络在预测中的应用 轨迹生成 概念 无人车是在许多物体间穿梭行驶&#xff0c;其中许多物体本身就是一直在移动的&#xff0c;比如像其他汽车、自行车、行人。无人车需要预测这些物体的行为&#xff0c;这样才能确保做出最佳决策。…

工作中对InheritableThreadLocal使用的思考

最近在工作中结合线程池使用 InheritableThreadLocal 出现了获取线程变量“错误”的问题&#xff0c;看了相关的文档和源码后在此记录。 1. 先说结论 InheritableThreadLocal 只有在父线程创建子线程时&#xff0c;在子线程中才能获取到父线程中的线程变量&#xff1b;当配合…

coding持续集成

先看看官网的一些操作提示 1、创建SSH密钥对 2、创建制品仓库 看完官网的介绍&#xff0c;持续集成需要提前准备好SSH凭证和制品仓库&#xff0c;下面将让我们动手开始吧 一、创建SSH密钥对 登录服务器控制台&#xff0c;创建 SSH 密钥对。获取私钥对后将其录入至 CODING 中…

Netty源码阅读(2)之——服务端源码梗概

上文我们把客户端源码梗概大致了解了一下&#xff0c;这样再了解服务端源码就轻松一点&#xff0c;我们将从服务端和客户端的区别着手去解析。 目录 区别 ④ ③ ① ⑤ 区别 ④ 客户端&#xff1a;.option(ChannelOption.TCP_NODELAY, true) 在TCP/IP协议中&#xff0c;无论…

贪心算法小结

A-金银岛 某天KID利用飞行器飞到了一个金银岛上&#xff0c;上面有许多珍贵的金属&#xff0c;KID虽然更喜欢各种宝石的艺术品&#xff0c;可是也不拒绝这样珍贵的金属。但是他只带着一个口袋&#xff0c;口袋至多只能装重量为w的物品。岛上金属有s个种类, 每种金属重量不同&am…

ffmpeg视频编解码 demo初探(一)(包含下载指定windows版本ffmpeg)分离视频文件中的视频流每一帧YUV图片

参考文章1&#xff1a;YUV数据流编码成H264 参考文章2&#xff1a;【FFmpeg编码实战】&#xff08;1&#xff09;将YUV420P图片集编码成H264视频文件 文章目录第一个项目&#xff1a;分离视频文件中的视频流每一张图片弯路步入正轨下载官方编译的ffmpeg4.3&#xff08;win64-g…