如何使用 FastAPI 返回 JSON 格式的数据？

Question

我在两者中都编写了具有相同功能的相同 API 应用程序快速API和烧瓶.但是，在返回 JSON 时，两个框架之间的数据格式不同。两者都使用相同的json 库，甚至使用相同的代码：

import json
from google.cloud import bigquery
bigquery_client = bigquery.Client()

@router.get('/report')
async def report(request: Request):
    response = get_clicks_impression(bigquery_client, source_id)
    return response

def get_user(client, source_id):
    try:
        query = """ SELECT * FROM ....."""
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                bigquery.ScalarQueryParameter("source_id", "STRING", source_id),
            ]
        )
        query_job = client.query(query, job_config=job_config)  # Wait for the job to complete.
        result = []
        for row in query_job:
            result.append(dict(row))
        json_obj = json.dumps(result, indent=4, sort_keys=True, default=str)

    except Exception as e:
        return str(e)

    return json_obj

返回的数据在烧瓶是字典：

  {
    "User": "fasdf",
    "date": "2022-09-21",
    "count": 205
  },
  {
    "User": "abd",
    "date": "2022-09-27",
    "count": 100
  }
]

而在快速API是字符串：

"[\n    {\n        \"User\": \"aaa\",\n        \"date\": \"2022-09-26\",\n        \"count\": 840,\n]"

我使用json.dumps() 的原因是date 不能被迭代。

Answer 1

如果你在返回对象之前序列化它——例如，使用json.dumps()，就像你的例子一样——对象最终会被序列化两次，因为 FastAPI 会自动序列化返回值。因此，您最终得到输出字符串的原因。请参阅下面的可用选项。

选项1

在首先使用jsonable_encoder 将数据转换为JSON 兼容数据（例如dict）后，您通常可以返回dict、list 等数据和FastAPI would automatically convert that return value into JSON。 jsonable_encoder 确保不可序列化的对象，例如datetime 对象，被转换为str。然后，在幕后，FastAPI 会将与 JSON 兼容的数据放入 JSONResponse 中，这将向客户端返回 application/json 编码响应。 JSONResponse，可以在 Starlette 的源代码 here 中看到，将使用 Python 标准 json.dumps() 序列化 dict（对于交替/更快的 JSON 编码器，请参阅 this answer）。

数据：

from datetime import date

d = [{'User': 'a', 'date': date.today(), 'count': 1},
        {'User': 'b', 'date':  date.today(), 'count': 2}]

API端点：

@app.get('/')
def main():
    return d

以上等价于：

from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder

@app.get('/')
def main():
    return JSONResponse(content=jsonable_encoder(d))

输出：

[
  {
    "User": "a",
    "date": "2022-10-20",
    "count": 1
  },
  {
    "User": "b",
    "date": "2022-10-20",
    "count": 2
  }
]

选项 2

如果出于任何原因（例如，尝试强制使用某些自定义 JSON 格式），您必须在返回对象之前对其进行序列化，那么您可以return a custom Response directly，如this answer 中所述。根据documentation：

当您直接返回 Response 时，其数据为不是验证， 转换（序列化），也没有自动记录。

此外，如here 所述：

FastAPI（实际上是 Starlette）会自动包含一个 内容长度标头。它还将包含一个 Content-Type 标头， 基于media_type 并为文本类型附加一个字符集。

因此，您还可以将media_type 设置为您期望数据的任何类型；在这种情况下，即application/json。下面给出示例。

注1：您的问题中显示的输出似乎是使用sort_keys=False 而不是sort_keys=True 生成的，如您的代码所示；因此，您在下面的输出中可能注意到的任何差异都是因为这个原因。

笔记2：此答案中发布的输出字符串是通过浏览器直接访问 API 端点的结果（即，通过在浏览器的地址栏中键入 URL，然后按 Enter 键）。如果您通过 Swagger UI 在/docs 测试端点，您会看到缩进不同（在此答案中提供的两个选项中）。这是由于 Swagger UI 如何格式化 application/json 响应。如果您还需要在 Swagger UI 上强制使用自定义缩进，则可以避免为 Response 指定 media_type。这将导致内容显示为文本，因为响应中将缺少 Content-Type 标头，因此，Swagger UI 无法识别数据的类型，以便对其进行格式化。

注3: 在json.dumps() 中将default 参数设置为str 可以序列化日期对象，否则如果未设置，您将得到：TypeError: Object of type date is not JSON serializable。 default 是一个函数，它被调用来处理无法序列化的对象。它应该返回对象的 JSON 可编码版本。在这种情况下，它是str，这意味着每个不可序列化的对象都将转换为字符串。如果您想以自定义方式序列化对象，您还可以使用自定义函数或 JSONEncoder 子类，如 demosntrated here。

注4：FastAPI/Starlette 的 Response 接受 content 参数作为 str 或 bytes 对象。如实现here 所示，如果您不传递bytes 对象，Starlette 将尝试使用content.encode(self.charset) 对其进行编码。因此，例如，如果您传递了dict，您将得到：AttributeError: 'dict' object has no attribute 'encode'。在下面的示例中，传递了一个 JSON str，稍后将其编码为 bytes（您也可以在将其传递给 Response 对象之前自行对其进行编码）。

API端点：

from fastapi import Response
import json

@app.get('/')
def main():
    json_str = json.dumps(d, indent=4, sort_keys=True, default=str)
    return Response(content=json_str, media_type='application/json')

输出：

[
    {
        "User": "a",
        "count": 1,
        "date": "2022-10-20"
    },
    {
        "User": "b",
        "count": 2,
        "date": "2022-10-20"
    }
]