1：开篇
1.在实践中检验真理 在实践中发展真理
陆陆续续的整理整个学习Fastapi框架一些小笔记的过程发现，自己其实对某些知识点掌握到实际应用的时候还是容易忘记，而且部分知识点也是难以理解和使用，熬了一段时间之后，也算是完整了公司内部的后台从flask迁移到Fastapi框架上。
再迁移的过程中深知，对于python异步的使用，还是需要多多实践，才能深入去理解和应用asyncio异步IO的使用，也累积了自己一点使用这个框架过程中一些使用小经验。
为了检验自己的知识掌握程度和系统化的进行学习和使用这个框架，另外不辜负关注我的公众号的粉丝们（虽然没多少个，但是仍然感激他们的不离不弃，毕竟我写的东西感觉太凌乱，太烂了~哈哈）之前说有没有实践，所以不管怎么样，还是把自己这些时间整理的脚手架搭建的一些知识点再进行系统的补充一下。
2.内容概要
相关的文章叙述主要通过实战叙述我们日常一些 API 构建过程一些流程，其实无非而已是开发-测试-部署三个范畴：
- 开发环境的搭建（docker环境搭建）
- API设计一些规范描述
- 对脚手架个功能点的整理
+ 项目配置文件处理
+ API日志记录处理
+ 异步redis缓存处理
+ 同步数据库整合使用
+ 异步数据库的整合和使用
+ 全局错误异常处理
+ 全局的Http请求响应报文的处理
+ 扩展第三方插件-限流器
+ 扩展第三方的插件-错误统计处理
+ 扩展-第三方插件-全局的认证JWT
+ 扩展-第三方插件-消息队列的整合
+ API版本的规划和处理
+ Api相关的单元测试和引入相关性能分析
+ 可能的话（还是带出K8S的应用（我还在学习中~））- 然后就是完整一个API服务的部署，这里主要是结合Docker和Drone进行相关部署
内容上，可能也会随时改变，但是整体可能还是会围绕上面几个概要点展开。
以上几点是大概接下来我的文章我自己会去整理的，因为我的表达能力有限，可能有些地方诉述会词不达意，希望海涵，也希望各位大佬批评指出。
PS:脚手架纯属个人经验的不断优化累积的沉淀结果，属于个人经验之谈，仅供参考！如有错误，还请各位大佬指正！
3.关于脚手架
3.1 序言
首先，经过之前一系列的基础知识的学习，大致上如何把这个框架应用到我们的自己的业务环境中的，就需要适当进行整合和封装一下，让它更加快捷方便自己的业务开放。
如何的去封装整合一个属于合适自己的脚手架，这个要看自己喜欢是没有的风格了吧，另外如果公司中也有相关的规范要求的话，那你只能根据公司的规范要求来设计。
但是总体上，我个人觉得一个脚手架其实是离不开几点要素，我自己的话，则封装一些常用几个功能（其实也是上面提到几个要点）：
- 支持 Swagger 接口文档生成（fastapi纯天然自带基因）
- 需要支持对应的日志收集-我这里使用loguru进行日志记录
- 扩展支持相关的JWT的接口鉴权验证
- 支持相关的接口的cors跨域支持
- 对中间件扩展的支持（但是fastapi中间件扩展如何涉及读取reques中的body等信息会有问题，需要考虑自己的需求）
- 如果你喜欢配置文件类型的读取的方式，还可以进行相关配置文件解析
- 全局的统一响应返回体（无论是错误还是正常的）
- 内部的日志的自定义traceid链路追踪(包含第三方的请求处理)
- 支持相关的rate 接口限流（基于redis）
- 支持全局的异常捕获，可以进行相关通知
- sentry异常的捕获上传处理
3.2 脚手架整体结构说明
因为如果项目是针对单一的项目，不是一个大型的项目的话，可以不需提取公共的模块出来，如果有有必要的公共的模块想共享使用的话，就可以提取公共的出来！我这里暂时不提炼出来。
以下是我的自己脚手架的整体的一个结构：
PS:纯属个人的组织方式，仅供参考
3.3计划基于脚手架之上会案例结构小示例:
- 使用异步的方式进行对接三地方接口（天气预报接口）
- 基于Vue制作一个简单的用户管理系统（使用别人的Vue模板是最快滴）
- 长远计划，弄一个数据自己的后台系统
2.搭建开始
一个脚手架的搭建首先从规划我们的项目结构开始（当然需要根据自己企业自身的而定咯）。
首先就是如下截图，是我自己规划的结构图：
规划好的项目结构之后，那接下里我们第一步就是开始先定义一个我们自己的一个fastapi的对象以及初始化这个基础的对象的时候一些配置信息读取。
因为之前使用flask，习惯了那种工厂模式方式来初始化我们的对象，所以这里也沿袭了那种风格来定义：
从上面的可以大概理解出我们的整个的应用启动的时候，需要哪些配件了！
2.1 全局配置文件说明
因为启动一个APP对象的时候，涉及到一些配置信息的时候，我们的需要统一去管理，为了更细致区分，甚至我自己划分了更明晰的配置功能：
比如上图所示的，有：
- 全局认证的配置信息
- 文档配置信息
- 数据库的配置信息
- redis的配置信息
PS:通常生产环境，根据个人需求吧，为了安全性，肯定是把一些重要信息通过写入环境变量来读取！可以结合一下读取环境变量的方式来解析对应的配置信息！
dotenv 这个库可以了解一下！
主要配置文件信息有：
auth_conf.py：
1 | python复制代码#!/usr/bin/evn python |
docs_conf.py(初始化我们的Fastapi对象的时候使用到)：
1 | ini复制代码from functools import lru_cache |
pgdb_conf.py：
1 | ini复制代码from functools import lru_cache |
redis_conf.py：
1 | python复制代码from functools import lru_cache |
以上是是所以关于配置信息说明:
2.2 App示例对象创建
一个Fastapi的实例就是一个应用服务，这个服务需要哪些配件，再启动的时候就需要配置好。所以有了以下规划：
主要内容就是：
创建一个APP对应的时候，顺便实例化我们的一些插件或导入的API服务等，一下是详细的方法的说明：
所以我们的展开的时候也会按上面的相关注册功能模块展开。
2.2.1 注册日志模块的处理
在一个应用中日志对我们的后续的异常定位是必不可缺的一部分，所以我们的需要对我们的相关的请求日志做好写入本地，便于后续回溯问题定位。
这里的日志模块处理，我使用的是loguru,这个库其实是挺不错的日志库，也支持异步的写入，所以再异步里的，感觉使用上应该是很理想的。
再定义日志模块之前，需要考虑的问题点：
- 日志存贮目录
- 日志记录格式
- 日志切割处理
那基于上述几个问题，我们把我们的日志记录，也进行一个插件化的方式来引入。
所以再我们的ext模块下，就有对应日志插件的处理：
- 定义日志：
loger_config.py 文件内容：
1 | python复制代码import time |
PS:注意点，再使用这个日志处理器的时候，下面句很关键避免多次的写入我们的日志，没有这个的话，会在记录日志的时候重复写入多次！
logger.configure(handlers=[{‘sink’: stdout, ‘format’: LOGURU_FORMAT}])
- 定义日志记录对象：
为什么需要这样的处理，因为之前几篇文章，我也有叙说过，关于fastapi在中间件处理日志的时候问题，使用中间件的方式记录日志的话，我们的无法对
request: Request 二次使用，而我的自己对日志需求就是如下示例：
1 | json复制代码2021-08-03 10:21:14:718 | process_id:24088 process_name:MainProcess | thread_id:2684 thread_name:MainThread | INFO | {"traceid": "DKVNm2JsxA2wVhY2hvFSQa", "trace_index": 1, "event_type": "request", "msg": {"useragent": {"os": "Windows 10", "browser": "QQ Browser 10.8.4405", "device": {"family": "Other", "brand": null, "model": null}}, "url": "/check", "method": "GET", "ip": "127.0.0.1", "params": {}, "ts": "2021-08-03 10:21:14"}} |
因为我的请求日志和记录响应的日志是分开记录，所以我上面处理方式和某大佬提出的是有点出路的。
基于我自己上面那种需求，所以我就分开记录，分开记录就需要一个 thread_id：所有有下面使用自定义ContextLogerRoute的实现：
文件名称：
1 | 复制代码contexr_logger_route.py |
文件内容：
1 | python复制代码from time import perf_counter |
有了上面的插件之后，那接下里就可以直接再app对象里面注册了.
有了上面仅仅是对我们的日志对象和我们日志信息的配置的初始化，下一步我们还需要对应我们的路由的日志也进行初始化：
给我们的路由添加自定义的路由实现，用于请求日志的记录：
有了上面的注册时候，如果在有需要记录日志的地方只需要使用：
1 | ini复制代码 |
验证示例如下：
从上面看我们可以把整个请求的链路日志给记录下来了！
2.2.2 注册全局异常捕获信息
全局异常处理器，对于统一异常的拦截捕获和统一响应报文，是非常关键滴！
2.2.2.1 异常类插件-扩展位置：
2.2.2.2 异常类插件- 注册方式：
思考，其实我们的所有的插件扩展思路无法就是把我们app对象注册到我们的扩展中，让他们可以在我们的扩展里面引用app对象相关调用，所以顺着这个思路其实我们扩展我们所谓的插件，这样就可以很方面定义自己需要东西了！
如我们的全局异常的对象主要使用的我们的app的异常拦截的，所以我们的可以自定义自己的一个异常的捕获类，来处理相关的所有的异常。
2.2.2.3 异常类插件- 类实现：
1 | python复制代码#!/usr/bin/evn python |
PS:上面实现其实可以使用两种方式，一个是定义函数的方式，然后通过add_exception_handler处理我们的异常；另一种其实就是直接装饰器的方式，为了清晰点我们的采取的第一种方案！
2.2.2.4 异常类插件- 验证：
验证函数故意引发异常：
结果：
异常引发捕获位置：
对应的引发异常响应报文定义：
2.2.2.5 补充全局响应报文定义）：
- 定义的位置：
默认其实是对JSON解析的有三种，我们为方便分开了三个，这个只显示一个！ - json_response.py
1 | ini复制代码 |
#!/usr/bin/evn python
-- coding: utf-8 --
“””
文件名称 : json_response
文件功能描述 : 功能描述
创建人 : 小钟同学
创建时间 : 2021/7/15
修改描述-2021/7/15:
“””
from typing import Any, Dict, Optional
自定义返回的错误的响应体信息
ORJSONResponse一依赖于：orjson
from fastapi.responses import JSONResponse
import time
from fastapi.encoders import jsonable_encoder
class ApiResponse(JSONResponse):
# 定义返回响应码–如果不指定的话则默认都是返回200
http_status_code = 200
# 默认成功
api_code = 0
# 默认Node.如果是必选的，去掉默认值即可
result: Optional[Dict[str, Any]] = None # 结果可以是{} 或 []
message = ‘成功’
success = True
timestamp = int(time.time() * 1000)
def __init__(self, success= None, http_status_code=None, api_code=None, result=None, message=None, **options):
if result:
self.result = result
if message:
self.message = message
if api_code:
self.api_code = api_code
if success != None:
self.success = success
if http_status_code:
self.http_status_code = http_status_code
# 返回内容体
body = dict(
message=self.message,
code=self.api_code,
success=self.success,
result=self.result,
timestamp=self.timestamp,
# 形如request="POST v1/client/register"
# request=request.method + ' ' + self.get_url_no_param()
)
# customize_headers = {
# # 'Access-Control-Allow-Origin': '*',
# # access-control-allow-methods: DELETE, GET, OPTIONS, PATCH, POST, PUT
# 'access-control-allow-methods': 'DELETE, GET, OPTIONS, PATCH, POST, PUT',
# 'Access-Control-Allow-Origin': '*',
# 'Access-Control-Allow-Headers': '*,X-Access-Token,School-Teacher-Token,school-teacher-token,T-Access-Token,x-access-token,Referer, Accept, Origin, User-Agent,X-Requested-With, Content-Type, X-File-Name',
# # 'Access-Control-Request-Headers': 'Content-Type,Access-Token',
# 'Content-Type': 'application/json;charset=UTF-8'
# }
# jsonable_encoder 处理不同字符串返回 比如时间戳 datatime类型的处理
super(ApiResponse, self).__init__(status_code=self.http_status_code, content=jsonable_encoder(body), **options)
# 这个render会自动调用，如果这里需要特殊的处理的话，可以重写这个地方
# def render(self, content: Any) -> bytes:
#
# return dict_to_json_ensure_ascii_indent(content)class BadrequestException(ApiResponse):
http_status_code = 400
api_code = 10031
result = None # 结果可以是{} 或 []
message = ‘错误的请求’
success = False
class LimiterResException(ApiResponse):
http_status_code = 429
api_code = 429
result = None # 结果可以是{} 或 []
message = ‘访问的速度过快’
success = False
class ParameterException(ApiResponse):
http_status_code = 400
result = {}
message = ‘参数校验错误,请检查提交的参数信息’
api_code = 10031
success = False
class UnauthorizedException(ApiResponse):
http_status_code = 401
result = {}
message = ‘未经许可授权’
api_code = 10032
success = False
class ForbiddenException(ApiResponse):
http_status_code = 403
result = {}
message = ‘失败！当前访问没有权限，或操作的数据没权限!’
api_code = 10033
success = False
class NotfoundException(ApiResponse):
http_status_code = 404
result = {}
message = ‘访问地址不存在’
api_code = 10034
success = False
class MethodnotallowedException(ApiResponse):
http_status_code = 405
result = {}
message = ‘不允许使用此方法提交访问’
api_code = 10034
success = False
class OtherException(ApiResponse):
http_status_code = 800
result = {}
message = ‘未知的其他HTTPEOOER异常’
api_code = 10034
success = False
class InternalErrorException(ApiResponse):
http_status_code = 500
result = {}
message = ‘程序员哥哥睡眠不足，系统崩溃了！’
api_code = 500
success = False
class InvalidTokenException(ApiResponse):
http_status_code = 401
api_code = 401
message = ‘很久没操作，令牌失效’
success = False
class ExpiredTokenException(ApiResponse):
http_status_code = 422
message = ‘很久没操作，令牌过期’
api_code = 10050
success = False
class FileTooLargeException(ApiResponse):
http_status_code = 413
api_code = 413
result = None # 结果可以是{} 或 []
message = ‘文件体积过大’
class FileTooManyException(ApiResponse):
http_status_code = 413
message = ‘文件数量过多’
api_code = 10120
result = None # 结果可以是{} 或 []
class FileExtensionException(ApiResponse):
http_status_code = 401
message = ‘文件扩展名不符合规范’
api_code = 10121
result = None # 结果可以是{} 或 []
class Success(ApiResponse):
http_status_code = 200
api_code = 200
result = None # 结果可以是{} 或 []
message = ‘自定义成功返回’
success = True
class Fail(ApiResponse):
http_status_code = 200
api_code = 200
result = None # 结果可以是{} 或 []
message = ‘自定义成功返回’
success = False
1 |
注意点：
上面的一个地方是引入jsonable_encoder
解决的事是部分的json数据类型的的问题！
2.2.3 全局配置跨域设置
这个跨域比较简单，这里不展开叙述，这里提一点就是可以针对某个地址ijnx跨域设置！通过设置：allow_origins来设置自持跨域的白名单。
2.2.4 注册全局中间件的注册
通常全局中间件处理一般对处理认证的，因为大部分的认证都是所有的接口的，所以为了方便，使用中间件的方式进行认证处理是最简单的方式。
这里我们使用全局认证的插件来说明：
2.2.4.1 全局中间件-认证中间件位置：
2.2.4.2 全局中间件-注册方式：
2.2.4.3 全局中间件-实现类：
1 | python复制代码#!/usr/bin/evn python |
2.2.4.3 全局中间件-认证验证：
新增一个接口，没加入白名单：
开启我们的中间件热的认证：
访问没加白的地址：
2.2.5 注册全局的启动和关闭事件
这个地方其实没什么需要特殊的全局处理，所以没什么内容，当然如果后续的如数据库的地方，就需要处理下，但是不是使用这种方式来处理了！而是也是插件的方式来处理！
所以这里没什么内容需要展开的，或者可以忽略！
2.2.6 注册全局第三方扩展插件实例
这里第三方的扩展的示例，可以根据自身的需要是否需要再这里进行注册实例化。
我这里的话其实可以选择实例化，比如实例化一个自定义实现了async_client，这样可以就可以全局再其他地址直接使用这个对象实例了！
2.2.6.1 插件 AsynClientSession 实例-位置：
2.2.6.2 插件 AsynClientSession 实例-注册：
2.2.6.3 插件 AsynClientSession 实例-类实现：
这个自定义实现，主要是增加第三方接口请求的时候相关的日志，
1 | python复制代码#!/usr/bin/evn python |
2.2.6.4 插件 AsynClientSession 实例-验证：
查看请求记录日志信息：
2.2.7 批量导入注册路由
因为我们的一个项目里面可能包含的路由比较多，如果一个一个的去
1 | scss复制代码app.include_router(router) |
我个人是不太喜欢这种方式！所以我自己参考了以前flask的模式，进行批量的导入注册，
也就是说，寻找某个模块下某个实例对象属性，进行统一的
1 | scss复制代码app.include_router(router) |
具体的实现方式如下步骤：
2.2.7.1 指定导入模块：
2.2.7.2 批量导入工具类-指定导入项目目录：
知道了要查询的目录，然后遍历下面有没有关于
1 | 复制代码bp |
实例属性，然后动态进行挂载：
1 | ini复制代码router = getattr(module, key_attribute) |
2.2.7.3 批量导入工具类-实现：
1 | python复制代码from fastapi import FastAPI, FastAPI |
关于里面使用到的
1 | javascript复制代码from apps.utils.modules_helper import find_modules, import_string |
主要是使用flask的代码：
1 | python复制代码import sys |
2.2.7.4 批量导入工具类-最终效果：
项目接口的定义：
接口定义示例：
不需要太多的干预处理直接的批量导入相关API接口定义：
以上就是关于路由批量导入的一些简要说明，后续有机会再针对这个定义API再展开一下！
2.3 插件异步redis示例扩展
我们的异步redis也是已插件的方式注册和实例化，主要的是以
上面这种方式引入的我们的app对象来接管注册对应的钩子函数，并在钩子函数处理相关事件处理。
2.3.1 插件定义位置：
2.3.1 插件类实现定义：
1 | python复制代码#!/usr/bin/evn python |
2.3.1 插件类初始化和验证：
我们再app启动的时候就示例这个redis异步客户端实例，并且进行相关一些验证测试：
3.总结
关于脚手架内容概要提到几个点，上面讲述的几乎已全部包含了！鉴于文章太长！后续如果可以的话，再继续展开！！！
以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！
结尾
END
简书：www.jianshu.com/u/d6960089b…
公众号：微信搜【小儿来一壶枸杞酒泡茶】
小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822
本文转载自: 掘金