FastAPI：如何在返回后记录POST方法的返回值？ [英] FastAPI: How to log the return value of a POST method after returning the response?

问题描述

我正在开发我的第一个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？&q；之后自动调用self.write_data，但如果我完全使用了错误的方法，也欢迎提出任何其他建议：)

推荐答案

您可以为此使用background task。后台任务将仅在"；发送后运行(根据Starlette documentation)。您可以定义一个任务函数在后台运行，用于写入日志数据，如下图所示：

def write_log_data():
    logger.write_data()

向route定义一个类型声明为BackgroundTasks的参数，在route内部，使用add_task()方法将(上述)任务函数传递给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

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

这篇关于FastAPI：如何在返回后记录POST方法的返回值？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！