How to initialize a global object or variable and reuse it in every FastAPI endpoint?

Question

I am having a class to send notifications. When being initialized, it involves making a connection to a notification server, which is time-consuming. I use a background task in FastAPI to send notifications, as I don't want to delay the response due to the notification. Below is the sample code:

file1.py

noticlient = NotificationClient()

@app.post("/{data}")
def send_msg(somemsg: str, background_tasks: BackgroundTasks):
    result = add_some_tasks(data, background_tasks, noticlient)
    return result

file2.py

def add_some_tasks(data, background_tasks: BackgroundTasks, noticlient):
    background_tasks.add_task(noticlient.send, param1, param2)
    result = some_operation
    return result

Here, the notification client is declared globally. I could have it initialized in file2.py, under add_some_tasks, but it would get initialized every time a request arrives, and that would require some time. Is there any way to use a middleware to re-use it every time a request arrives, so that it doesn't need to be initialized every time?

Or, another approach might be to initialize notification in class definition:

file1.py

class childFastApi(FastAPI):
    noticlient = NotificationClient()

app = childFastApi()

@app.post("/{data}")
def send_msg(somemsg: str, background_tasks: BackgroundTasks):
    result = add_some_tasks(data, background_tasks, app.noticlient)
    return result

Answer 1

Option 1

You could store the custom class object to the app instance, which could be used to store arbitrary extra state using the generic app.state attribute, as demonstrated here, as well as here and here. To access the app.state dictionary—and subsequently the variable/object that you had stored to state—outside the main application file (for instance, accessing it from a routers submodule that uses APIRouter), you could use the Request object, as demonstrated in this answer (i.e., using request.app.state). You could either use a startup event (as shown here) to initialize the object, but since it is now deprecated (and might be removed from future versions), you could instead use a lifespan handler function.

Example

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' Run at startup
        Initialize the Client and add it to app.state
    '''
    app.state.n_client = NotificationClient()
    yield
    ''' Run on shutdown
        Close the connection
        Clear variables and release the resources
    '''
    app.state.n_client.close()


app = FastAPI(lifespan=lifespan)


@app.get('/')
async def main(request: Request):
    n_client = request.app.state.n_client
    # ...

Option 2

Since the introduction of Starlette's lifespan handler, which, similar to startup and shutdown event handlers, allows one to define code that needs to run before the application starts up, and/or when the application is shutting down, one could also define objects to be accessible from request.state (which is recommended over app.state). As per Starlette's documentation:

The lifespan has the concept of state, which is a dictionary that can be used to share the objects between the lifespan, and the requests.

The state received on the requests is a shallow copy of the state received on the lifespan handler.

Hence, after instantiating the class object in the lifespan handler, you could then add it to the state dictionary, and later access it within the various endpoints—even those defined in APIRouters outside the main application file—using request.state.

Example

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' Run at startup
        Initialize the Client and add it to request.state
    '''
    n_client = NotificationClient()
    yield {'n_client': n_client}
    ''' Run on shutdown
        Close the connection
        Clear variables and release the resources
    '''
    n_client.close()


app = FastAPI(lifespan=lifespan)


@app.get('/')
async def main(request: Request):
    n_client = request.state.n_client
    # ...

Important Note

Even though app.state and request.state are not meant to be used for storing variables/objects that are expected to be modified by every request, there might be cases, however, such as in a simple (single-user) application, that might be more convenient for the developer to utilize the state as such (even though not really recommended) than setting up some persistent storage (Note, however, that in a multi-user application, where multiple users could have concurrent or parallel access to the same variable/object, using the state for such purposes might not be a wise choice, and one should rather look into thread-safe solutions, as mentioned in this answer, for instance).

In such cases, i.e., when using requst.state to store a variable/object that is expected to be modified by the various requests arriving at the API, one should store a str/int/float/etc. variable into dict or list objects instead of storing them directly to the state. The reason is simply because, as mentioned earlier, "the state received on the requests is a shallow copy of the state received on the lifespan handler" (the relevant uvicorn implementation part may be found here). A shallow copy, as described in Python's documentation, "constructs a new compound object and then inserts references into it to the objects found in the original". A shallow copy is only relevant for compound objects, i.e., objects that contain other objects, such as list, dict or class instances. Hence, str/int/float/etc. objects stored directly to the state cannot be changed; thus, any changes made to them in the new (copied) object would not be reflected in the original object, in contrast to compound objects described earlier. For the sake of completeness, it should be noted that changes made to str/int/float/etc. objects stored directly to app.state would actually be applied (compared to request.state), as app.state is the original object itself and not a shallow copy that is received on every request, as in the case of request.state.

A simple example demonstrating what has been explained above can be found below.

Example

app.py

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield {"data": {"val": 1}}
    #yield {"val": 1}  # changes to `val` would not take effect globally


app = FastAPI(lifespan=lifespan)


@app.get('/set')
async def set(request: Request):
    #request.state.val = 2  # changes to `val` would not take effect globally
    request.state.data["val"] = 2
    return request.state


@app.get("/get")
async def get(request: Request):
    return request.state

test.py

import httpx

r = httpx.get(url="http://127.0.0.1:8000/get")
print(r.json())
#{'_state': {'data': {'val': 1}}}

r = httpx.get(url="http://127.0.0.1:8000/set")
print(r.json())
#{'_state': {'data': {'val': 2}}}

r = httpx.get(url="http://127.0.0.1:8000/get")
print(r.json())
#{'_state': {'data': {'val': 2}}}