How to read body as any valid json?

Question

I haven't found the docs for that use case. How can I get the request body, ensure it's a valid JSON (any valid JSON, including numbers, string, booleans, and nulls, not only objects and arrays) and get the actual JSON. Using Pydantic forces the JSON to have a specific structure.

Answer 1

You can find nearly everything inside the Request object

You are able to get request body with request.json(), which will give you the parsed JSON as dictionary.

from fastapi import Request, FastAPI

@app.post("/dummypath")
async def get_body(request: Request):
    return await request.json()

If you want access the body as string, you can use request.body()

Answer 2

If you are confident that the incoming data is "a valid JSON", you can create a simple type annotation structure to receive the arbitrary JSON data.

from fastapi import FastAPI
from typing import Any, Dict, AnyStr, List, Union

app = FastAPI()

JSONObject = Dict[AnyStr, Any]
JSONArray = List[Any]
JSONStructure = Union[JSONArray, JSONObject]


@app.post("/")
async def root(arbitrary_json: JSONStructure = None):
    return {"received_data": arbitrary_json}

Examples

1. JSON object

curl -X POST "http://0.0.0.0:6022/" -H  "accept: application/json" -H  "Content-Type: application/json" -d "{\"test_key\":\"test_val\"}"

Response:

{
  "received_data": {
    "test_key": "test_val"
  }
}

2. JSON array

curl -X POST "http://0.0.0.0:6022/" -H  "accept: application/json" -H  "Content-Type: application/json" -d "[\"foo\",\"bar\"]"

Response:

{
  "received_data": [
    "foo",
    "bar"
  ]
}

---

If you are not sure about the content type of the incoming data, better to parse the request body.

It can be done as,

from fastapi import FastAPI, Request

app = FastAPI()


@app.post("/")
async def root(request: Request):
    return {"received_request_body": await request.body()}

The advantage of this method is that the body will contain any kind of data, JSON, form-data, multipart-form-data, etc.

Answer 3

The accepted answer is valid as well, but FastAPI provides a built-in way to do that - check the Singular values in body section in docs.

A parameter with the default Body gets all the payload that doesn't match passed Pydantic-typed parameters (the whole payload in our case) and converts it to the appropriate Python type. In case of invalid JSON, a standard validation error would be produced.

from typing import Any
from fastapi import Body, FastAPI

app = FastAPI()


@app.post('/test')
async def update_item(
        payload: Any = Body(None)
):
    return payload

UPD: Note on first Body positional argument (default) - None here makes request body optional, ... (Ellipsis) - marks it as required (passing nothing will actually keep it required). Read more in the Required with Ellipsis docs section

Also, this solution works for JSON containing only null, true, false, any string, any number.

Answer 4

For those of you using BaseModel and want to have a JSON field, you can import Json from pydantic

from fastapi import FastAPI
from pydantic import BaseModel, Json, Field


app = FastAPI()


class MockEndpoint(BaseModel):
    endpoint: str = Field(description="API endpoint to mock")
    response: Json = Field(description="Example response of the endpoint")


@app.get("/")
async def root():
    return {"message": "Hello World"}


@app.post("/mock")
async def mock_request(mock_endpoint: MockEndpoint):
    return mock_endpoint

Answer 5

This is also another method to get any form of Json as your input

@app.post("/dict/")
async def post_dict(data: Dict[str, Any]):
    return data

But I guess its not the cleanest way of doing things

Answer 6

This is an example to print the content of a Request, it will print the json body (if it is json parsable) otherwise just print the raw bytes of the body.

async def print_request(request):
        print(f'request header       : {dict(request.headers.items())}' )
        print(f'request query params : {dict(request.query_params.items())}')  
        try : 
            print(f'request json         : {await request.json()}')
        except Exception as err:
            # could not parse json
            print(f'request body         : {await request.body()}')
    
    
    @app.post("/printREQUEST")
    async def create_file(request: Request):
        try:
            await print_request(request)
            return {"status": "OK"}
        except Exception as err:
            logging.error(f'could not print REQUEST: {err}')
            return {"status": "ERR"}

Answer 7

FastAPI has a JSON encoder.

There are some cases where you might need to convert a data type (like a Pydantic model) to something compatible with JSON

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
import simplejson as json

class SubmitGeneral(BaseModel):
    controllerIPaddress: str
    readerIPaddress: str
    ntpServer: str

@app.post("/submitGeneral")
async def submitGeneral(data: SubmitGeneral):
    data = jsonable_encoder(data)
    #data = json.loads(data.json()) # same as above line
    
    print(f"data = {json.dumps(data)}")

    # you have to access the properties with brackets, not by dot notation
    query = f"update LocalPLC set ControllerIpAddress = '{data['controllerIPaddress']}', ReaderIPAddress = '{data['readerIPaddress']}'"

    return {"status": "OK"}