对于带有 json 列的 sqlite db，fastapi 响应的格式不正确

Posted 2023-03-22

【中文标题】对于带有 json 列的 sqlite db，fastapi 响应的格式不正确

【英文标题】：fastapi response not formatted correctly for sqlite db with a json column

【发布时间】：2021-10-16 08:44:39

【问题描述】：

我有一个带有 sqlite 的快速 api 应用程序，我试图将输出作为有效的 json。 sqlite数据库中的一列是存储在Text列中的列表，另一列在Text列中具有json数据。

下面的代码示例

database = Database("sqlite:///db/database.sqlite")

app = FastAPI()

@app.get("/flow_json")
async def get_data(select: str='*'):
    query = query_formatter(table='api_flow_json',select=select)

    logger.info(query)
    results = await database.fetch_all(query=query)
    print(results)
    # this result is a list of tuples which i can confirm output stated below
    
    return  results

打印的元组列表

[('182', 'ABC', 'response_name', '[["ABC","DEF","GHI"]]', 'GHI', '"metadata":"contentId":"ABC"', '2', 'false', '39', '72', 'true')]

下面的sqlite db行示例

"id","customer_name","response_name","entities","abstract","json_col","revision","disabled","customer_id","id2","auth"
182,"ABC","response_name","[[""ABC"",""DEF"",""GHI""]]","GHI","""metadata"":""contentId"":""ABC""",2,false,39,72,true

使用 http 调用的结果

["id":"182","customer_name":"ABC","response_name":"response_name","entities":"[[\"ABC\",\"DEF\",\"GHI\"]]","abstract":"GHI","json_col":"\"metadata\":\"contentId\":\"ABC\"","revision":"2","disabled":"false","customer_id":"39","id2":"72","auth":"true"]

预期结果

["id":"182","customer_name":"ABC","response_name":"response_name","entities":[["ABC","DEF","GHI"]],"abstract":"GHI","json_col":metadata:contentId:ABC,"revision":"2","disabled":"false","customer_id":"39","id2":"72","auth":"true"]

我做了什么：

在获得元组列表后，将列表转换为对 json 更友好 尝试了 sqlite 的 json1 扩展，但不起作用。 我知道这将涉及数据库响应后的格式设置，但无法确定返回给客户端的格式。

【参考方案1】：

扩展 this 答案我将使用 Json 数据类型为 entities 和 json_col：

class ApiFlowJson(BaseModel):
    id: int
    customer_name: str
    response_name: str
    entities: Json
    abstract: str
    json_col: Json
    revision: int
    disabled: bool
    customer_id: int
    id2: int
    auth: bool

    class Config:
        orm_mode = True

样本

from typing import List

import sqlalchemy
from databases import Database
from fastapi import FastAPI
from pydantic import BaseModel, Json


DATABASE_URL = "sqlite:///test.db"


app = FastAPI()

# database
database = Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

api_flow_json = sqlalchemy.Table(
    "api_flow_json",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("customer_name", sqlalchemy.String),
    sqlalchemy.Column("response_name", sqlalchemy.String),
    sqlalchemy.Column("entities", sqlalchemy.JSON),
    sqlalchemy.Column("abstract", sqlalchemy.String),
    sqlalchemy.Column("json_col", sqlalchemy.JSON),
    sqlalchemy.Column("revision", sqlalchemy.Integer),
    sqlalchemy.Column("disabled", sqlalchemy.Boolean),
    sqlalchemy.Column("customer_id", sqlalchemy.Integer),
    sqlalchemy.Column("id2", sqlalchemy.Integer),
    sqlalchemy.Column("auth", sqlalchemy.Boolean),
)

engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args="check_same_thread": False
)

metadata.create_all(engine)

# pydantic

class ApiFlowJson(BaseModel):
    id: int
    customer_name: str
    response_name: str
    entities: Json
    abstract: str
    json_col: Json
    revision: int
    disabled: bool
    customer_id: int
    id2: int
    auth: bool

    class Config:
        orm_mode = True


# events

@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


# route handlers

@app.get("/")
def home():
    return "Hello, World!"


@app.get("/seed")
async def seed():
    query = api_flow_json.insert().values(
        customer_name="ABC",
        response_name="response_name",
        entities='["ABC","DEF","GHI"]',
        abstract="GHI",
        json_col='"metadata":"contentId":"ABC"',
        revision=2,
        disabled=False,
        customer_id=39,
        id2=72,
        auth=True,
    )
    record_id = await database.execute(query)
    return "id": record_id


@app.get("/get", response_model=List[ApiFlowJson])
async def get_data():
    query = api_flow_json.select()
    return await database.fetch_all(query)

【讨论】：

这行得通，但你能解释一下这里发生了什么吗，我确实理解了所有部分，但是 sqlalchemy 引擎在这里有什么影响？和使用的json。我会很感激这个

您可以忽略 sqlalchemy 位。我假设您只是使用普通 SQL 来设置数据库并创建表，对吗？对于您的代码，请使用 sqlite.org/json1.html 为 entities 和 json_col 启用 JSON 字段。

我使用 vanilla sql 并使用 CSV 导入数据。最主要的是 pydantic Json 也为我的数据库工作，现在一切都完美呈现。

【参考方案2】：

您应该使用 pydantic BaseModel 进行响应：

from pydantic import BaseModel
# Possible additional code

class Metadata(BaseModel):
    contentId: str

class JsonCol(BaseModel):
    metadata: Metadata

class ApiFlowJson(BaseModel):
    id: int
    customer_name: str
    response_name: str
    entities: List[str]
    abstract: str
    json_col: JsonCol
    revision: int
    disabled: bool
    customer_id: int
    id2: int
    auth: bool

    class Config:
        orm_mode = True


@app.get("/flow_json", response_model=List[ApiFlowJson])
async def get_data(select: str='*'):
    # your code

更详细的解释可以在这里找到：https://fastapi.tiangolo.com/tutorial/response-model/

如果你不想进入 Pydantic，另一种方法是直接返回响应：

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

@app.get("/flow_json")
async def get_data(select: str='*'):
    # your code
    json_compatible_data = jsonable_encoder(results)
    return JSONResponse(content=json_compatible_data)

有关直接响应的更多详细信息，请参见此处：https://fastapi.tiangolo.com/advanced/response-directly/

注意：代码没有执行和测试，所以你不仅应该复制粘贴它，还要检查它

【讨论】：

我收到一个错误 pydantic.error_wrappers.ValidationError: 2 个 ApiFlowCatalog 响应的验证错误 -> 0 -> 实体值不是有效列表 (type=type_error.list) 响应 -> 0 -> json_col值不是有效的字典 (type=type_error.dict)

我认为这是因为实体是字符串内的列表，对于 json_col 也是如此，它是字符串内的 json

是的，你可能是对的。将其（entities 和 response）都更改为字符串。

但是我如何将响应转换为 json，这是有效的，比如 list 是一个带有字符串封装的列表？

我没看懂你最后的评论，请详细说明