How to use async starlette.request functions in a synchronous endpoint

[kbakk]

First check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.
• After submitting this, I commit to one of:
  • Read open issues with questions until I find 2 issues where I can help someone and add a comment to help there.
  • I already hit the "watch" button in this repository to receive notifications and I commit to help at least 2 people that ask questions in the future.
  • Implement a Pull Request for a confirmed bug.

Example

Here's a self-contained, minimal, reproducible, example with my use case:

from fastapi import FastAPI
from starlette.requests import Request
import uvicorn

app = FastAPI()

@app.post("/reports/", response_model=None)
def submit_report(request: Request):
    # access to reproduce issue
    body = request.body()

    ## workaround:
    # import asyncio
    # body = asyncio.run(request.body())

    # ... do stuff with body ...

uvicorn.run(app, host="0.0.0.0", port=8000)

# output:
# INFO:     Started server process [97531]
# INFO:     Waiting for application startup.
# INFO:     Application startup complete.
# INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
# __main__:5: RuntimeWarning: coroutine 'Request.body' was never awaited
# RuntimeWarning: Enable tracemalloc to get the object allocation traceback
# INFO:     127.0.0.1:54472 - "POST /reports/ HTTP/1.1" 200 OK

Description

I need to access the raw request body for processing, so I'm using starlette.request.Request as suggested in the docs (and in #2460).

Run above in one terminal, and this in another curl -XPOST localhost:8000/reports/.

As seen seen above (uvicorn log output), getting RuntimeWarning: coroutine 'Request.body' was never awaited. This is because the body method is async (ref.).

Is asyncio.run(request.body()) the best (only) workaround here, if I don't want my route to be async?

Environment

• OS: MacOS Catalina 10.15.7
• FastAPI Version: 0.63.0
• Python version: Python 3.7.6

[Kludex]

Hi @kbakk!

No, those aren't the only options. Well, if you want to stick with the choices you've made:

from fastapi import FastAPI, Request
import uvicorn

app = FastAPI()

@app.post("/reports/")
async def submit_report(request: Request):
    body = await request.body()


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

You need to use coroutines (async), which can be awaited (you can use await).

But if you want a more FastAPIonic way, you can validate your body with a Pydantic schema, e.g.:

from fastapi import FastAPI, Request
from pydantic import BaseModel
import uvicorn

app = FastAPI()

class MyBody(BaseModel):
    my_word: str
    my_num: int

@app.post("/reports/")
async def submit_report(my_body: MyBody):
    ...


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

[raphaelauv]

I don't want my route to be async

If your endpoint is not async stalette will run it in an external threadpool , and very few case make this a better choice

https://fastapi.tiangolo.com/async/#path-operation-functions

if the answer of @Kludex is alright , @kbakk dont forget to close the issue , thank you

[kbakk]

Thanks for the answers here. However, it doesn't seem to address my issue.

In the first example it's using async, which I can't use (my dependencies are synchronous). The second example is not really allowing me to access the body - only a body that's provided as JSON or form data (if I understand correctly).

In my use case, I want users to upload plain text files using something like curl -XPOST localhost:8000/reports/ --data-binary @path/to/file.txt (but I suppose this question is valid in case the body is in any other format as well - or if one would access request.stream).

If your endpoint is not async stalette will run it in an external threadpool , and very few case make this a better choice

I'm not too worried about the performance penalty of running it synchronously.

[raphaelauv]

my dependencies are synchronous

You can call synchronous from asynchronous code ... and since you say :

I'm not too worried about the performance penalty of running it synchronously.

So why don't you just do what proposed @Kludex ?

def super_synchronous_code():
    sleep(1)

@app.post("/reports/")
async def submit_report(request: Request):
    body = await request.body()
    super_synchronous_code()

[kbakk]

    sleep(1)

Won't executing that in an async function block execution of other calls?

Why don't you just do what proposed @Kludex ?

As mentioned, I have dependencies (SQLAlchemy in my case) that's synchronous, and while I know about e.g. databases to make it work well asynchronously, I don't feel like introducing additional dependency and complexity I don't understand.

Per the docs it recommends to use synchronous endpoints:

If you are using a third party library that communicates with something (a database, an API, the file system, etc) and doesn't have support for using await, (this is currently the case for most database libraries), then declare your path operation functions as normally, with just def

I see that my initial workaround (while it doesn't feel "clean") is probably good enough for me. Thanks for the input here. :)

I'll leave it open for a little while longer, in case there's other ways around this.

[raphaelauv]

Won't executing that in an async function block execution of other calls?

ANY sync code block execution of other calls ( so only IO async operations or async.sleep are really releasing the execution to the loop )

The advise of the documentation to use only DEF is when you only have sync code to execute. ( maybe this advise could be a little more explained in the doc )

In your case to "limit" the impact of your sync function on the main async loop of fastAPI, you can do this :

from starlette.concurrency import run_in_threadpool

def synchronous_code_than_could_be_long_blocking(arg:int)-> int:
    print(arg)
    sleep(10) # any long synchronous call ( like SQLAlchemy call )
    return arg

@app.post("/reports/")
async def submit_report(request: Request):
    body = await request.body()
    rst = await run_in_threadpool(synchronous_code_than_could_be_long_blocking, 1378473)

[falkben]

I think using a pydantic model to represent accepted body data is probably the best, since it will also self document your route and perform validation (as @Kludex suggested).

Buf if you really wanted to leave it open ended as you have it currently, AND you had some sort of blocking code in this endpoint that prevents you from awaiting the body in the route itself, another way around this issue is to use a async dependency to pull out the body.

you can have async dependencies on non-async routes.

so it could be:

from fastapi import Depends, FastAPI, Request

app = FastAPI()


async def get_body(request: Request):
    # or request.json() if you are sure the body is coming in that way
    return await request.body()


@app.post("/reports/", response_model=None)
def submit_report(body=Depends(get_body)):
    return body

[kbakk]

It doesn't seem like a good idea to call synchronous blocking code in an async function as suggested. It will block all processing request. Yes, it's as much blocking in a synchronous endpoint, but in that case it would probably run it with several threads.

run_in_threadpool does seem like an option for preventing synchronous code to block, and yes - then I can do await request.body(). I'm still leaning towards asyncio.run(request.body()) though, as I can keep the entire application synchronous (which means I'll need to run it in several threads/processes in production so I'll pay a small performance penalty).

Thanks for the help here 👍

[kbakk]

Since there has been suggested some ways to address this, I'll close this issue, considering it as answered.

[kbakk]

After further testing with asyncio.run(request.body()), I noticed it in some cases would throw

RuntimeError: Task <Task pending coro=<Request.body() running at venv/lib/python3.7/site-packages/starlette/requests.py:219> cb=[run_until_complete.<locals>.<lambda>()]> got Future <Future pending> attached to a different loop

This approach by @falkben seems to work though. 👍

[perkfly]

This is a definitely valid issue. I have a use case that in a legacy system the request body is so complex that it contains 30+ nested fields and not well documented. Defining a pydantic model for the request would not be very viable solution in a short time. Directly accessing the request.json in a sync manner would be very useful. I really hope that FastAPI could add a wrapper around starlette.requests.Request to provide a sync version of request.json/body etc.

[ecly]

@yifeikong I agree and have encountered a use case myself, that is not trivially solved by the approach described by @falkben. I want to run some computationally expensive function from a sync endpoint, and be able to occasionally check if we should stop further computation, in the event that a client disconnected.

To do this, I wanted to use is_disconnected as a "cancellation function" and pass it to my expensive function, but this was not trivially possible as request.is_disconnected is async and if implemented naively, we run into "got Future attached to a different loop" as mentioned by @kbakk.

The "solution" I ended up with was:

async def is_disconnected(request: Request):
    main_loop = asyncio.get_event_loop()

    def is_disconnected_sync():
        future = asyncio.run_coroutine_threadsafe(request.is_disconnected(), main_loop)
        try:
            return future.result(0.01)
        except asyncio.exceptions.TimeoutError:
            return False

    return is_disconnected_sync

@router.post("/expensive")
def translate(
    is_disconnected_func = Depends(is_disconnected),
):
    try:
        return expensive_computation(cancel_function=is_disconnected_func)
    except CancelledComputation:
        return JSONResponse(status_code=499)  

It's not that nice, so would very much appreciate a native solution, but it does the job for my current use case.