fastapi踩坑记

语言: CN / TW / HK

theme: condensed-night-purple highlight: atom-one-dark


持续创作,加速成长!这是我参与「掘金日新计划 · 6 月更文挑战」的第30天,点击查看活动详情

大家好~我是小方,欢迎大家关注笋货测试笔记体完记得俾个like

踩坑-脱坑

组内维护着一个造数平台,用的是fastapi为web框架,为了体验更好,做了一些自定义异常处理,手动捕获异常和自动捕获错误,捕获的信息通过日志和企微机器人发送异常信息或者报错信息,但是自动捕获错误,无法捕获body,这让我很头疼!!!详细可看TesterHome原贴:https://testerhome.com/topics/28909 帖子放上去之后,都没能解决;问了公司的测开,涉及到知识盲区,也没能解决!之前这个坑,也就暂时放下来了~ python 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。

详细说明可看官方文档:https://fastapi.tiangolo.com/tutorial/middleware/?h=middleware

下面是无敌哥pity平台代码,详细可看GitHub ```python 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:https://github.com/tiangolo/fastapi/issues/394 https://stackoverflow.com/questions/61358669/raise-exception-in-python-fastapi-middleware

自定义路由类-APIRoute

在某些情况下,您可能希望覆盖Request和APIRoute类使用的逻辑。特别是,这可能是中间件中逻辑的一个很好的替代方案。例如,如果您想在应用程序处理请求正文之前读取或操作请求正文。 详细说明可看官方文档:https://fastapi.tiangolo.com/advanced/custom-request-and-route ```python 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:https://github.com/tiangolo/fastapi/issues/1216 https://github.com/tiangolo/fastapi/issues/2750

总结

今天介绍了fastapi踩坑之路,希望大家以后不要踩这个坑~遇到报错时,网上找不到资料,第一找无敌哥(无敌卷魔、肝帝!),第二看官方文档,第三看官方的issue!总能解决你的问题!勇敢牛牛,不怕困难!

fastapi学习资料:https://fastapi.tiangolo.com/