对于带有 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"}]

我尝试了什么：

1. 获得元组列表后，将列表转换为更加 json 友好
2. 尝试了 sqlite 的 json1 扩展，但不起作用。
3. 我知道这将涉及数据库响应后的格式化，但无法找出返回给客户端的格式。

Answer 1

Expanding on this answer I would use the Json datatype for entities and 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

Sample

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)