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

ghz 1years ago ⋅ 2122 views

Question

I am having a class to send notifications. When being initialised, 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 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, notification client is declared globally. I could have it initialised in file2.py under add_some_tasks, but it would get initialised 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 initialised every time.

or Approach two: Initialize notification in class def

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

Option 1

You could store the custom class object to the app instance, which allows you to store arbitrary extra state using the generic the app.state attribute, as demonstrated here, as well as here and here. To access the app.state attribute, and subsequently the object, outside the main file (for instance, from a routers submodule that uses APIRouter), you could use the [Request](https://fastapi.tiangolo.com/advanced/using-request- directly/#use-the-request-object-directly) object, as demonstrated in this answer (i.e., using request.app.state). You could either use a startup event (as shown here) to initialise the object, but since it is now deprecated (and might be removed in future versions), you could instead use a lifespan function.

Example

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' Run at startup
        Initialise 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 events, allows one to define code that needs to run before the application starts up, or when the application is shutting down, one could also define objects to be accesible from the request.state. As per [Starlette's documentation](https://www.starlette.io/lifespan/#lifespan- state):

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, when instantiating the class object in the lifespan handler, you could then add it to the dictionary (i.e., the state), and access it within an endpoint using request.state.

Example

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' Run at startup
        Initialise 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
    # ...
#fastapi