The FastAPI application I started working on, uses several services, which I want to initialize only once, when the application starts and then use the methods of this object in different places.
It can be a cloud service or any other heavy class.
Possible ways is to do it with Lazy loading and with Singlenton pattern, but I am looking for better approach for FastAPI.
Another possible way, is to use Depends class and to cache it, but its usage makes sense only with route methods, not with other regular methods which are called from route methods.
Example:
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
async def non_route_function(commons: dict = Depends(common_parameters)):
print(commons) # returns `Depends(common_parameters)`
@router.get('/test')
async def test_endpoint(commons: dict = Depends(common_parameters)):
print(commons) # returns correct dict
await non_route_function()
return {'success': True}
There can be also used @app.on_event("startup") event to initialize heavy class there, but have no idea how to make this initialized object accessible from every place, without using singleton.
Another ugly way is also to save initialized objects into @app( and then get this app from requests, but then you have to pass request into each non-route function.
All of the ways I have described are either ugly, uncovenient, non-pythonic or worse practice, we also don't have here thread locals and proxy objects like in flask, so what is the best approach for such kind of problem I have described above?
Thanks!
It's usually a good idea to initialize the heavy objects before launching the FastAPI application. That way you're done with initialization when the application starts listening for connections (and is made available by the load balancer).
You can set up these dependencies and do any initialization in the same location as you set up your app and main routers, since they're a part of the application as well. I usually expose the heavy object through a light weight service that exposes useful endpoints to the controllers themselves, and the service object is then injected through Depends.
Exactly how you want to perform the initialization will depend on what other requirements you have in the application - for example if you're planning to re-use the infrastructure in cli tools or use them in cron as well.
This is the way I've been doing it in a few projects, and so far it has worked out fine and kept code changes located in their own vicinities.
Simulated heavy class in heavylifting/heavy.py with from .heavy import HeavyLifter in __init__.py:
import time
class HeavyLifter:
def __init__(self, initial):
self.initial = initial
time.sleep(self.initial)
def do_stuff(self):
return 'we did stuff'
A skeleton project created in a module named foo (heavylifting lives under foo/heavylifting for now to make sense of the imports below):
foo/app.py
from fastapi import FastAPI, APIRouter
from .heavylifting import HeavyLifter
heavy = HeavyLifter(initial=3)
from .views import api_router
app = FastAPI()
app.include_router(api_router)
foo/services.py
The service layer in the application; the services are the operations and services that the application exposes to controllers, handling business logic and other co-related activities. If a service needs access to heavy, it adds a Depends requirement on that service.
class HeavyService:
def __init__(self, heavy):
self.heavy = heavy
def operation_that_requires_heavy(self):
return self.heavy.do_stuff()
class OtherService:
def __init__(self, heavy_service: HeavyService):
self.heavy_service = heavy_service
def other_operation(self):
return self.heavy_service.operation_that_requires_heavy()
foo/app_services.py
This exposes the services defined to the application as dependency lightweight injections. Since the services only attach their dependencies and gets returned, they're quickly created for a request and then discarded afterwards.
from .app import heavy
from .services import HeavyService, OtherService
from fastapi import Depends
async def get_heavy_service():
return HeavyService(heavy=heavy)
async def get_other_service_that_uses_heavy(heavy_service: HeavyService = Depends(get_heavy_service)):
return OtherService(heavy_service=heavy_service)
foo/views.py
Example of an exposed endpoint to make FastAPI actually serve something and test the whole service + heavy chain:
from fastapi import APIRouter, Depends
from .services import OtherService
from .app_services import get_other_service_that_uses_heavy
api_router = APIRouter()
@api_router.get('/')
async def index(other_service: OtherService = Depends(get_other_service_that_uses_heavy)):
return {'hello world': other_service.other_operation()}
main.py
The application entrypoint. Could live in app.py as well.
from fooweb.app import app
if __name__ == '__main__':
import uvicorn
uvicorn.run('fooweb.app:app', host='0.0.0.0', port=7272, reload=True)
This way the heavy client gets initialized on startup, and uvicorn starts serving requests when everything is live. Depending on how the heavy client is implemented it might need to pool and recreate sockets if they can get disconnected for inactivity (as most database libraries offer).
I'm not sure if the example is easy enough to follow, or that if it serves what you need, but hopefully it'll at least get you a bit further.