FastAPI：如何在返回响应后记录 POST 方法的返回值？答案

【问题标题】：How to log the return value of a POST method after returning the response?FastAPI：如何在返回响应后记录 POST 方法的返回值？
【发布时间】：2022-08-02 19:24:23
【问题描述】：

所以我正在开发我的第一个 REST API，如果我错过了一些基本的东西，请提前道歉。无论如何，我有一个函数从另一台服务器接收一个 json 请求，处理它（根据数据进行预测），然后返回另一个带有结果的 json。我想在服务器的本地磁盘上记录到此端点的所有请求及其结果，用于评估目的和重新训练模型。但是，为了尽量减少返回结果给用户的延迟，我想先返回响应数据，然后再写入本地磁盘。如何正确地做到这一点对我来说并不明显，因为 FastAPI 范式要求 POST 方法的结果是修饰函数的返回值，所以我想对数据做的任何事情都必须完成前它被退回。

下面是一个最小的工作示例，我认为这是我迄今为止最接近正确的尝试，使用带有 log 装饰器的自定义对象 - 我的想法是将结果作为类属性分配给日志对象，然后使用另一种方法将其写入磁盘，但我不知道如何确保调用该函数后get_data 每次。

import json
import uvicorn
from fastapi import FastAPI, Request
from functools import wraps
from pydantic import BaseModel

class Blob(BaseModel):
    id: int
    x: float

def crunch_numbers(data: Blob) -> dict:
    # does some stuff
    return {\'foo\': \'bar\'}

class PostResponseLogger:

    def __init__(self) -> None:
        self.post_result = None

    def log(self, func, *args, **kwargs):
        @wraps(func)
        def func_to_log(*args, **kwargs):
            post_result = func(*args, **kwargs)
            self.post_result = post_result

            # how can this be done outside of this function ???
            self.write_data()

            return post_result
        return func_to_log

    def write_data(self):
        if self.post_result:
            with open(\'output.json\', \'w\') as f:
                    json.dump(self.post_result, f)

def main():
    app = FastAPI()
    logger = PostResponseLogger()

    @app.post(\'/get_data/\')
    @logger.log
    def get_data(input_json: dict, request: Request):
        result = crunch_numbers(input_json)
        return result

    uvicorn.run(app=app)

if __name__ == \'__main__\':
    main()

基本上，我的问题归结为：\“在PostResponseLogger 类中，有没有办法在每次调用self.log 后自动调用self.write_data？\”，但如果我完全使用了错误的方法，也欢迎任何其他建议:)

你可能想看看this answer。
我可能遗漏了一些东西，但我认为那里没有任何东西可以解决我的问题 - 据我所知，FastAPI 中间件仍然只能处理响应前退货

标签： python-3.x fastapi python-decorators

【解决方案1】：

为此，您可以使用Background Task。后台任务“将仅在发送响应后运行”（请参阅Starlette documentation）。 “这对于需要在请求之后发生的操作很有用，但客户端实际上不必在接收响应之前等待操作完成”（请参阅FastAPI documentation）。

您可以定义一个任务函数在后台运行，用于写入日志数据，如下所示：

def write_log_data():
    logger.write_data()

然后，导入BackgroundTasks 并在端点中定义一个参数，其类型声明为BackgroundTasks。在您的端点内部，使用方法.add_task() 将您的任务函数（即write_log_data，如上定义）传递给background_tasks 对象：

from fastapi import BackgroundTasks

@app.post('/get_data/')
@logger.log
def get_data(input_json: dict, request: Request, background_tasks: BackgroundTasks):
    result = crunch_numbers(input_json)
    background_tasks.add_task(write_log_data)
    return result

如果使用中间件来捕获和记录响应数据，则可以应用相同的原理，如this answer 中所述。作为将来的参考，如果您（或任何人）需要使用 async/await 语法并遇到并发问题（在执行一些繁重的后台计算时），请查看文档 here 以及 @987654326 @ 和 this 回答。

【讨论】：

啊，酷，没有意识到这一点。谢谢！