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
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("https://stackoverflow.com/")
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:
The
lifespan
has the concept ofstate
, 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("https://stackoverflow.com/")
async def main(request: Request):
n_client = request.state.n_client
# ...