对于具有 json 列的 sqlite 数据库，fastapi 响应格式不正确

Question

我有一个带有 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 数据库行示例

"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 扩展，但不起作用。
我知道这将涉及数据库响应后的格式化，但无法弄清楚 return 给客户端的格式。

Answer 1

您应该使用 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，可以直接 return 回复：

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/

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

Answer 2

为 entities 和 json_col 扩展 answer I would use the Json 数据类型：

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)

对于具有 json 列的 sqlite 数据库，fastapi 响应格式不正确

fastapi response not formatted correctly for sqlite db with a json column

python

sqlite

json

python-3.x

fastapi