fastapi踩坑记

持续创作，加快成长！这是我参加「日新方案 6 月更文应战」的第30天，点击查看活动概况

我们好~我是小方，欢迎我们关注笋货测试笔记体完记住俾个like呀

踩坑-脱坑

组内维护着一个造数渠道，用的是fastapi为web框架，为了体会更好，做了一些自定义反常处理，手动捕获反常和主动捕获过错，捕获的信息经过日志和企微机器人发送反常信息或者报错信息，但是主动捕获过错，无法捕获body，这让我很头疼！！！具体可看TesterHome原贴：testerhome.com/topics/2890…
帖子放上去之后，都没能处理；问了公司的测开，涉及到知识盲区，也没能处理！之前这个坑，也就暂时放下来了~

File "C:\Users\junjie\AppData\Local\Programs\Python\Python37\lib\site-packages\starlette\requests.py", line 167, in empty_receive
    raise RuntimeError("Receive channel has not been made available")
RuntimeError: Receive channel has not been made available

从头跳坑

最近在看无敌哥的pity渠道，用的是异步处理，看看香不香！！！拉了代码下来跑，跑到运转用例时，就报500过错，Response都没有回来，发现无敌哥把主动捕获过错的处理注释掉了，看到这段代码，想起之前的坑，赶紧敲击一下无敌哥，无敌哥马上去搜材料，真不愧是无敌卷魔，肝帝！

填坑

无敌哥刮遍了整个搜索引擎和fastapi的issues，找到了两个处理方案，无敌卷魔牛逼！

自定义中间件-Middleware

在FastAPI使用中使用中间件。
中间件实际上是一个函数，在每个request处理之前被调用，同时又在每个response回来之前被调用。

1. 首先接收拜访过来的request。
2. 然后针对request或其他功用执行自定义逻辑。
3. 传递request给使用程序继续处理。
4. 接收使用所发生的response。
5. 然后针对response或其他功用执行自定义逻辑。
6. 回来response。

具体阐明可看官方文档：fastapi.tiangolo.com/tutorial/mi…

下面是无敌哥pity渠道代码，具体可看GitHub

async def set_body(request: Request, body: bytes):
    async def receive() -> Message:
        return {"type": "http.request", "body": body}
    request._receive = receive
async def get_body(request: Request) -> bytes:
    body = await request.body()
    await set_body(request, body)
    return body
@pity.middleware("http")
async def errors_handling(request: Request, call_next):
    body = await request.body()
    try:
        await set_body(request, await request.body())
        return await call_next(request)
    except Exception as exc:
        return JSONResponse(
            status_code=status.HTTP_200_OK,
            content=jsonable_encoder({
                "code": 110,
                "msg": str(exc),
                "request_data": body,
            })
        )

参阅issues：github.com/tiangolo/fa…
stackoverflow.com/questions/6…

自定义路由类-APIRoute

在某些情况下，您可能期望掩盖Request和APIRoute类使用的逻辑。特别是，这可能是中间件中逻辑的一个很好的代替方案。例如，如果您想在使用程序处理恳求正文之前读取或操作恳求正文。
具体阐明可看官方文档：fastapi.tiangolo.com/advanced/cu…

class ErrorRouter(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()
        async def custom_route_handler(request: Request) -> Union[Response]:
            try:
                return await original_route_handler(request)
            except Exception as exc:
                log_msg = f"造数渠道捕获到系统过错：恳求途径:{request.url.path}\n"
                params = request.query_params
                if params:
                    log_msg += f"途径参数：{params}\n"
                boby = await request.body()
                if boby:
                    body = json.dumps(json.loads(boby),ensure_ascii=False)
                    log_msg +=f"恳求参数：{body}\n"
                if isinstance(exc, NormalException):
                    return JSONResponse(
                        status_code=status.HTTP_200_OK,
                        content={
                            "responseCode": Status.SYSTEM_EXCEPTION.get_code(),
                            "responseMsg": exc.errorMsg
                        },
                    )
                elif isinstance(exc, RequestValidationError):
                    message = ""
                    for error in exc.errors():
                        message += str(error.get('loc')[-1]) + ":" + str(error.get("msg")) + ","
                    return JSONResponse(
                        status_code=status.HTTP_200_OK,
                        content=jsonable_encoder({
                            "responseCode": Status.PARAM_ILLEGAL.get_code(),
                            "responseMsg": Status.PARAM_ILLEGAL.get_msg() + message[:-1]
                        })
                    )
                log_msg +=f"过错信息：{str(exc.args[0])}"
                mylog.error(log_msg)
                if PlatConfig.SWITCH == 1:
                    WeGroupChatBot.send_text(log_msg)
                return JSONResponse(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    content=jsonable_encoder({
                        "responseCode": Status.FAIL.get_code(),
                        "responseMsg": Status.FAIL.get_msg(),
                        "errorMsg":str(exc.args[0])
                    },
                    ))
        return custom_route_handler
def APIRouter():
    router = AppRouter()
    router.route_class = ErrorRouter
    return router

一致处理之后，再经过类型判别Exception，回来不同的Response~
注意：用了自定义过错路由，就不能再用 @app.exception_handler否则会重复捕获！！！

参阅issues：github.com/tiangolo/fa…
github.com/tiangolo/fa…

总结

今天介绍了fastapi踩坑之路，期望我们以后不要踩这个坑~遇到报错时，网上找不到材料，第一找无敌哥(无敌卷魔、肝帝！)，第二看官方文档，第三看官方的issue！总能处理你的问题！勇敢牛牛，不怕困难！

fastapi学习材料：fastapi.tiangolo.com/