How to universally handle and parse requests with any content type in FastAPI?

67 views Asked by At

I am developing an API Gateway service using FastAPI and httpx as the request proxy library. My application has a single HTTP endpoint function that handles all incoming requests:

@main_router.api_route('/{path:path}', methods=["GET", "POST", "PUT", "PATCH", "DELETE"])

I expect to receive various request types based on the content type, such as "application/json", "application/x-www-form-urlencoded", "multipart/form-data", among others (including raw body and file content). Currently, I extract the body content using the following function:

async def get_request_body(request: starlette.requests.Request):
    content_type = request.headers.get('content-type')
    if request.method in ['GET', 'DELETE'] or not content_type:
        return None
    if "application/json" in content_type:
        body = await request.json()
    elif "application/x-www-form-urlencoded" in content_type:
        form_data = await request.form()
        body = dict(form_data)
    elif "multipart/form-data" in content_type:
        form_data = await request.form()
        body = {key: value for key, value in form_data.items()}
    else:
        body = await request.body()
    return body

Then, I map the data, json, files, content variables with body content based on content type:

def get_request_data(
        content_type: str,
        body: Any,
) -> Tuple[Dict[str, Any] | None, Dict[str, Any] | None, Dict[str, Any] | None, bytes | None]:
    if not content_type:
        return None, None, None, None
    if 'application/json' in content_type:
        return None, ujson.dumps(body), None, None
    elif 'application/x-www-form-urlencoded' in content_type:
        return body, None, None, None
    elif 'multipart/form-data' in content_type:
        return None, None, body, None
    else:
        return None, None, None, body

Finally, I send the request using:

    data, json, files, content = get_request_data(content_type, body)
    transport = httpx.AsyncHTTPTransport(retries=2)
    async with httpx.AsyncClient(
            timeout=timeout,
            transport=transport,
    ) as client:
        response = await client.request(
            method=request.method,
            url=url,
            headers=modify_initial_request_headers(request.headers),
            params=request.query_params.multi_items(),
            data=data,
            json=json,
            files=files,
            content=content,
        )
        return response

This approach works well with all the requests I've tested so far, but I think this solution is not universal and somewhat inelegant. How can I improve it?

I have implemented a function to handle incoming requests of various content types and extract their body content accordingly. My expectation is that this function should correctly parse the body of the request based on its content type, whether it's JSON, form data, multipart data, etc.

0

There are 0 answers