Fastapi: [QUESTION] Check if the request context is available

[raghavio]

Is there a way to check if the request variable is available to use?
I'm coming from Flask and my codebase gets run from the API and inside RQ Workers (background queue).
Inside RQ Workers, the SQLAlchemy db session is maintained per worker and for API it's at the request level.
To get the session I have a function that looks something like this:

def get_session():
    if has_request_context():
        # Get it from the request. In case of Fast this will be probably be Depends(request.state.db)
    elif has_worker_context():
        # Pick it from a local variable

To run a query we simply call the above function to get the session.

How can I achieve something like this in Fastapi?

[dmontagu]

It depends on how you have things implemented.

In general, even if it is somehow possible, I think it may force an awkward design if you try to use the same get_session function to get a session inside both your fastapi endpoints and inside your workers.

I would recommend you write functions that accept a session as an argument, and then use a fastapi Depends to inject the session in your endpoints, and use an alternative approach to get a session in your workers.

This could be as simple as calling the fastapi endpoint functions while specifying a non-default value for the session (so that the Depends is not passed to the function) in the workers. Or it could mean having your fastapi endpoints basically just being wrappers around a single set of functions either the api or workers can call, if some minor amount of api-specific preparation is required.

I think it might be hard to give a better answer than that without a little more detail about the api and worker design.

[dmontagu]

If you really want to go down this route, you could maybe do something along the following lines:

def get_fastapi_request(request: Request):
    return request

def get_session(request: Request = Depends(get_fastapi_request)) -> Session:
    if isinstance(request, Request):
        # fastapi injected the value, so you know you are in a "request context"
        return request.state.db
    else:
        assert isinstance(request, fastapi.params.Depends)  # (don't need to perform this)
        # default value was passed, so in worker context
        ...

But again, I would encourage you to think if there is a way to design this that avoids this check.

[raghavio]

Thanks for this.

--
We have 130k lines of code and doing the refactor of injecting the session through function parameters would be too much of a work.

The get_session is simply a helper function that picks the session depending on where the code was run from. Which can be 3 places. From the api request, worker and python console or jupyter environment. It allows to run the exact same code without any pre-requisites of starting an app or instantiating some objects.

It works well for us in Flask and I'd like to stick to it unless there's something very wrong with this design being used with Fast.

[dmontagu]

I understand if it is due to legacy reasons.

I take it this means you are not using fastapi's dependency injection to get access to the request?

One way you might be able to handle the request-access in a route could be through the use of context vars. You could do

from contextvars import ContextVar
from typing import Optional

from fastapi import APIRouter, FastAPI, Depends
from starlette.requests import Request

fastapi_request_var: ContextVar[Optional[Request]] = ContextVar('fastapi_request', default=None)

def set_fastapi_request(request: Request):
    fastapi_request_var.set(request)

api_router = APIRouter()  # add all your endpoints to this router

app = FastAPI()
app.include_router(api_router, dependencies=[Depends(set_fastapi_request)])

def get_session():
    fastapi_request = fastapi_request_var.get()
    if fastapi_request is not None:
        return fastapi_request.state.db
    # Not inside a request context
    ...

Note 1: I haven't ever used context vars, or tested the above code, but based on the linked docs it seems like the above would be the way to accomplish it 😅.

Note 2: This could lead to some debugging challenges if, for example, you try to access the fastapi_request prior to it being set. For example, I don't know if context vars are carried across threads (e.g., if starlette's run_in_threadpool is called). (But that's probably going to be a risk with any design that doesn't pass the request directly.)

Also, be aware that a new Request instance is created during each middleware (this is related to how starlette is implemented). The state should be passed along properly, but I could imagine running into some surprising behavior if you aren't aware of this.

[tiangolo]

In short, as @dmontagu says, you can use context vars to create a "global" variable that stores a single session and then close it after sending the response. You need to use Python 3.7 or above, and yes, context vars are preserved when using the thread pool (normal def functions).

And make sure to close the session after the request and start a new one for the next.

---

I imagine you're using Flask-SQLAlchemy or similar.

That uses thread locals to store the session. That is based on the assumption that each request is handled by a single thread.

In FastAPI (or any async tool) everything is run "in turns" on the same thread, in an "event loop". And when you run blocking code / normal def functions, FastAPI runs them in a thread pool, during the same request.

So, a thread local could be shared by multiple requests (maybe even simultaneously) and when using normal def, functions and dependencies, they would get their own session, because they are run in their own thread.

That's what is solved by context vars. They are, as with thread locals, "global" variables that are actually different in each case. But in context vars they are associated to each "turn" in the event loop, instead of each thread.

[github-actions[bot]]

Assuming the original issue was solved, it will be automatically closed now. But feel free to add more comments or create new issues.

[1xuan]

I understand if it is due to legacy reasons.

I take it this means you are not using fastapi's dependency injection to get access to the request?

One way you might be able to handle the request-access in a route could be through the use of context vars. You could do

from contextvars import ContextVar
from typing import Optional

from fastapi import APIRouter, FastAPI, Depends
from starlette.requests import Request

fastapi_request_var: ContextVar[Optional[Request]] = ContextVar('fastapi_request', default=None)

def set_fastapi_request(request: Request):
    fastapi_request_var.set(request)

api_router = APIRouter()  # add all your endpoints to this router

app = FastAPI()
app.include_router(api_router, dependencies=[Depends(set_fastapi_request)])

def get_session():
    fastapi_request = fastapi_request_var.get()
    if fastapi_request is not None:
        return fastapi_request.state.db
    # Not inside a request context
    ...

Note 1: I haven't ever used context vars, or tested the above code, but based on the linked docs it seems like the above would be the way to accomplish it .

Note 2: This could lead to some debugging challenges if, for example, you try to access the fastapi_request prior to it being set. For example, I don't know if context vars are carried across threads (e.g., if starlette's run_in_threadpool is called). (But that's probably going to be a risk with any design that doesn't pass the request directly.)

Also, be aware that a new Request instance is created during each middleware (this is related to how starlette is implemented). The state should be passed along properly, but I could imagine running into some surprising behavior if you aren't aware of this.

The code don't work for def! The reason is those function run in different 'event'!