Optimal way to initialize heavy services only once in FastAPI

Question

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!

Homologate

Any reason why you can't do something like result_from_non_route_function = Depends(non_route_function) in your route signature when you need access to the result? Additionally, you can wrap it in an lru_cache decorator (as shown with the config dependency in the FastAPI reference, iirc). – Workhorse

Thanks for your comment @MatsLindh, but what if route method calls second, and second method calls third, where we need that dependency? I have to pass Depends stuff on each function, it's super inconvenient for readability and maintenance. Also, it is not caching mechanism I am worrying about, it's about proper architectural solution on application side. – Homologate

Well, then it depends on where you need the reference to your service. Some sort of dependency should be injected into your controller since you need to decide what action the controller should take. In that case you can inject the dependent service (if that dependency depends on something else) in the function you use to create the dependency - so that the service layer wraps the operations you want to perform on your data and exposes those relevant operations to your controller. – Workhorse

Thanks again @MatsLindh, understand what you mean, but Dependenices generally as the Depends class in Fastapi exists for completely different reasons, not to initialize heavy services, but to make your modules more abstrat dependents. I just mentioned what limits FastAPI Depends has, to emphasize that it doesn't fixes my problem. And my problem is not to use Depends with some weird ways, but to initialize heavy services conveniently. – Homologate

I see; thanks for expanding. This is just my experience: In those cases we've usually initialized the heavy service in the same location as were we set up the app object - they're both relevant to the startup of the application. We then wrap it in a service class as necessary to hide away the actual import and implementation if we need to change it later, but so far this has worked fine for us. We then use this service class together with a Depends hierarchy that populates the relevant services and arguments as needed by the routes (the heavy class is initialized on application startup). – Workhorse

The heavy initialization itself would depend on the requirements of the module itself, but is usually done with a helper function in the module (of the heavy service) or as a class or static method on a helper class. I prefer to avoid Singletons as they're not testing friendly and imposes arbitrary limitations that might turn up as issues later (if we suddenly need to talk to two different instances of the same cloud service, for example). – Workhorse

Agree! @Workhorse can you please show me an simple example of what you described? I mean the heavy class initialization and then service of this class. I just want to see how you accessing from service class on the initialized object which is in the same file as app. You can show me the code in answer, instead of `comment. – Homologate

Sorry about the delay; I had to find some time to carve out a more complete example to show how I'd solve it. Seems to work fine with my examples and mirrors how I've been solving similar issues for certain backend services. – Workhorse

Answer 1

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.

Workhorse

Thank you @Workhorse so much for such a great example and comprehensive answer! it is more clear than to add extra params into @app and thne to try to fetch it in another place. You really helped me! – Homologate

Declaring the HeavyLifter object as a module-level global is a bit icky. It will make it impossible to import the app module without doing all that heavyweight work. FastAPI startup and shutdown events are a little better, especially since they give you an opportunity to clean up gracefully when the application stops. (Not that they don't have their own ickiness.) – Insalubrious

An alternative to startup and shutdown events is to use an async context manager that gets passed as the lifespan parameter to FastAPI() if your python version is recent enough. Note that that is the same page as the "startup and shutdown events" page in Maxpm's comment. – Gowan

Also note that the FastAPI documentation is not versioned, and that lifespan parameter was only added two weeks ago. – Gowan

What if the heavy service requires cleanup? – Scowl

@Scowl Then it depends on when that cleanup should run - whether it's at the end of each request, or when the server terminates – Workhorse

Answer 2

I don't know any way which does not rely on a singleton variable. I think that's acceptable, as FastAPI also uses the app as a singleton.

I'm reusing MatsLindh's answer but show how to use lifespan events for tying the expensive service lifetime to our app lifetime. That also enables async preparation and clean up.

I'll not introduce an app-specific layer and use the Annotated helper to reduce boiler-plate.

foo/heavylifting.py

The HeavyLifter is our expensive service, entirely unaware of FastAPI. It requires async preparation and clean up meaning that __init__() is inexpensive.

import asyncio

class HeavyLifter:
    def __init__(self, initial: int) -> None:
        self._initial = initial

    async def prepare(self) -> None:
        await asyncio.sleep(self._initial)

    async def clean_up(self) -> None:
        await asyncio.sleep(self._initial)
    
    def do_stuff(self) -> str:
        return "we did stuff"

foo/app.py

Next to our app singleton, there is also a heavy singleton. However, we use lifespan events to initialise it asynchronously only during app startup, and similarly run clean up as the app is shutting down.

from contextlib import asynccontextmanager
from fastapi import FastAPI, APIRouter
from typing import AsyncGenerator
from .heavylifting import HeavyLifter

heavy = HeavyLifter(initial=3)

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    global heavy # not necessary, as heavy is only read not written to
    try:
        await heavy.prepare()

        # Serve FastAPI app
        yield
    finally:
        await heavy.clean_up()

from .views import api_router

app = FastAPI(lifespan=lifespan)
app.include_router(api_router)

foo/dependencies.py

This file contains simplified dependency definitions - but it could just as well be part of app.py.

With Annotated we define a type alias for reuse across dependent functions. The singleton is loaded via an anonymous lambda function.

from fastapi import Depends
from typing import Annotated, TypeAlias
from .app import heavy
from .heavylifting import HeavyLifter

HeavyLifterDep: TypeAlias = Annotated[HeavyLifter, Depends(lambda: heavy)]

foo/views.py

For our path operation function, we use the type alias prepared in dependencies and will get the persistent, heavy service object with full type checker support.

from fastapi import APIRouter, Depends
from .dependencies import HeavyLifterDep

api_router = APIRouter()

@api_router.get('/')
async def index(heavy_lifter: HeavyLifterDep):
    return {'hello world': heavy_lifter.do_stuff()}

---

I was really surprised that FastAPI does not have a simpler, native way of having globally persistent dependencies. Maybe there is still a better way.

Donor