返回响应后如何记录 POST 方法的返回值？

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

下面是一个最小的工作示例，我认为这是迄今为止我最接近的尝试，使用带有log装饰器 - 我的想法只是将结果作为类属性分配给日志对象，然后使用另一种方法将其写入磁盘，但我不知道如何确保调用该函数after 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.write_data每次致电后self.log？”，但如果我完全使用了错误的方法，也欢迎任何其他建议。

你可以有一个Background Task https://fastapi.tiangolo.com/tutorial/background-tasks/为了这个目的。后台任务“仅在发送响应后运行”（请参阅星光文档 https://www.starlette.io/background/）。 “这对于需要在请求之后发生的操作很有用，但客户端在收到响应之前实际上不必等待操作完成”（请参阅FastAPI文档 https://fastapi.tiangolo.com/tutorial/background-tasks/).

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

def write_log_data():
    logger.write_data()

然后，导入BackgroundTasks并在端点中定义一个参数，其类型声明为BackgroundTasks。在端点内部，传递您的任务函数（即write_log_data，如上所定义）到background_tasks对象与方法.add_task()， 如下所示。有关如何使用的更多详细信息BackgroundTasks可以找到这个答案 https://stackoverflow.com/a/76280152/17865804.

from fastapi import BackgroundTasks

@app.post('/get_data/')
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

如果使用中间件来捕获和记录响应数据，则可以应用相同的原理，如中所述这个答案 https://stackoverflow.com/a/70899261/17865804，或自定义APIRoute类，如所示这个答案 https://stackoverflow.com/a/73464007/17865804.

供将来参考，如果您（或任何人）需要使用async/await语法，并遇到并发问题（例如event loop https://docs.python.org/3/library/asyncio-eventloop.html被阻止）在执行一些繁重的后台计算时，请看看这个答案 https://stackoverflow.com/a/71517830/17865804，这解释了定义端点或后台任务函数之间的区别async def and def（简要地，async def端点/后台任务将在event loop， 然而def函数将在外部线程池中运行，然后awaited），并提供了在此类函数中运行阻塞 I/O 密集型或 CPU 密集型操作时的解决方案。