How to return separate JSON responses using FastAPI?

I am not sure if this is part of OpenAPI standard. I am trying to develop an API server to replace an existing one, which is not open source and vendor is gone. One particular challenge I am facing is it returns multiple JSON objects without enclosing them either in a list or array. For example, it returns the following 3 JSON objects as they are, in separate lines:

{"items": 10}
{"order": "shelf", "amount": 100}
{"id": 100, "date": "2022-01-01", "status": "X"}

Not in a list format () or in array [].

For example, the code below returns all 3 objects in an array:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    data_1 = {"items": 10}
    data_2 = {"order": "shelf", "amount": 100}
    data_3 = {"id": 100, "date": "2022-01-01", "status": "X"}
    return data_1, data_2, data_3

Can anyone help me to get this done with FastAPI?

Solution

Option 1

You could return a custom Response directly, as demonstrated in this answer, as well as in Option 2 of this answer.

Example

from fastapi import FastAPI, Response
import json

app = FastAPI()


def to_json(d):
    return json.dumps(d, default=str)
    
  
@app.get('/')
async def main():
    data_1 = {'items': 10}
    data_2 = {'order': 'shelf', 'amount': 100}
    data_3 = {'id': 100, 'date': '2022-01-01', 'status': 'X'}
    json_str = '\n'.join([to_json(data_1), to_json(data_2), to_json(data_3)])
    return Response(json_str, media_type='application/json')

Option 2

You could use a StreamingResponse, as shown here and here. You might also find this and this helpful. If the generator function performs some blocking operations that would block the event loop, then you could define the gen() function below with a normal def instead of async def, and FastAPI will use iterate_in_threadpool() to run the generator in a separate thread that will then be awaited. Have a look at the linked answers above for more details.

Example

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()


@app.get('/')
async def main():
    data_1 = {'items': 10}
    data_2 = {'order': 'shelf', 'amount': 100}
    data_3 = {'id': 100, 'date': '2022-01-01', 'status': 'X'}
    
    async def gen():
        for d in [data_1, data_2, data_3]:
            yield json.dumps(d, default=str) + '\n'

    return StreamingResponse(gen(), media_type='application/json')

Option 3

As mentioned in the comments section above, one could also return a dictionary of dict (JSON) objects. However, using this solution, adding a line break between the objects would not be feasible.

Example

from fastapi import FastAPI, Response

app = FastAPI()


@app.get('/')
async def main():
    data_1 = {'items': 10}
    data_2 = {'order': 'shelf', 'amount': 100}
    data_3 = {'id': 100, 'date': '2022-01-01', 'status': 'X'}
    return {1: data_1, 2: data_2, 3: data_3}

Note

Although in Options 1 & 2 the media_type is set to application/json, the returned object would not be a valid JSON, as JSON strings do not allow real newlines (only escaped ones, i.e., \\n)—see this answer as well. Hence, in Swagger UI autodocs at /docs, you may come across the following message when testing the endpoint: can't parse JSON. Raw result:. If you would like to avoid getting that message, then you could set the media_type to text/plain instead (for Option 2, where StreamingResponse is used, you might want to disable MIME Sniffing when using text/plain; alternatively, you could use text/event-stream—see this answer for more details.).