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

问题描述：

我正在开发我的第一个 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.write_data`self.log`

解决方案 1：

你可以Background Task为此目的创建一个后台任务。后台任务“仅在响应发送后运行”（参见Starlette 文档）。“这对于需要在请求之后执行的操作很有用，但客户端实际上不必等待操作完成后再接收响应”（参见FastAPI 文档）。

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

def write_log_data():
    logger.write_data()

然后，在端点中导入BackgroundTasks并定义一个参数，其类型声明为BackgroundTasks。在端点内部，将任务函数（即 ，write_log_data如上所定义）传递给background_tasks具有 方法的对象.add_task()，如下所示。有关如何使用的更多详细信息，BackgroundTasks请参阅此答案。

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

如果使用中间件来捕获和记录响应数据（如此答案中所述）或自定义类（如此答案APIRoute中所示），则可以应用相同的原理。

为了将来的参考，如果您（或任何人）需要使用 async/await语法，并且在执行一些繁重的后台计算时遇到并发问题（例如event loop被阻塞），请查看此答案async def，它解释了使用和定义端点或后台任务函数之间的区别def（简而言之，async def端点和后台任务函数都将直接在中运行event loop，而普通def函数将在与外部线程池分开的线程中运行，然后被await编辑），并在运行阻塞 I/O 绑定或 CPU 绑定操作时提供解决方案在这些函数中。