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

user_7586544 0
Viewed 1082

I have a fast api app with sqlite, I am trying to get an output as json which is valid. One of the columns in sqlite database is a list stored in Text column and another column has json data in Text column.

code sample below

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

List of tuples printed

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

sqlite db row example below

"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

result using http call

[{"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"}]

expected result

[{"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"}]

What did I try:

  1. transforming list to be more json friendly after I get the list of tuples
  2. tried the json1 extension for sqlite but doesn't work.
  3. I know that this will involve a formatting after response from database but can't figure out the formatting to return to client.
fastapi python-3.x sqlite python json
2 Answers

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)
user_17994080

You should use pydantic BaseModel for your response:

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

A more detailed explanation can be found here: https://fastapi.tiangolo.com/tutorial/response-model/

Other way, if you don't want go into Pydantic, is to return response directly:

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)

More detailed on direct response can be found here: https://fastapi.tiangolo.com/advanced/response-directly/

NOTE: the code is not executed and tested, so you should not only copy-paste it, but also check it

user_43448860
Related