Add Starlette lifespan handler implementation

[ZipFile]

trafficstars

Starlette recently added cool new feature called lifespan in v0.26.0 and FastAPI followed the suit in v0.93.0. This makes it possible to get rid of good chunk of initialization spaghetti in starlette-based apps (making container a single entry point in the application), + reduces amount of the pain in the ass scheduling background tasks in the event loop of the ASGI app.

Example

myapp/container.py

from dependency_injector.containers import DeclarativeContainer
from dependency_injector.ext.starlette import Lifespan
from dependency_injector.providers import Factory, Resource, Self, Singleton
from fastapi import FastAPI


def init():
  print("Inittializing resources")
  yield
  print("Cleaning up resources")


class Container(DeclarativeContainer):
    __self__ = Self()
    lifespan = Singleton(Lifespan, __self__)
    app = Factory(FastAPI, lifespan=lifespan)
    init = Resource(init)

myapp/asgi.py:

from .container import Container

container = Container()
app = container.app()

run.sh:

uvicorn myapp.asgi:app

[ZipFile]

Looks like project is still alive. Rebased with latest master.

[VladyslavHl]

Great! Integrating DI Resources with FastAPI lifespan is something me and colleagues will investigate soon, good to see the potential of universal solution.

IMO it would be benefitial to make this feature more flexible, allowing user to chose what resources to initialize in lifespan as you may want to init some resources at other moment

From the top of my head Lifespan can take individual resources to be initialized/shutdown on lifespan events, so the usage could be:

class Container(DeclarativeContainer):
    init = Resource(init)
    other_resource = Resource(init_other_resource)   # to be intialized not on startup event
    lifespan = Singleton(Lifespan, init, ... )
    app = Factory(FastAPI, lifespan=lifespan)
    ...

also preserving the option to pass __self__ to init all resources of container as it is implemented now.

This leads to question how to not initialize these resources when calling container.init_resources(), maybe the its sufficient to just init them explicily by calling container.other_resource.init() when needed.

maybe I'm complicating, let me know wdyt

[ZipFile]

@VladyslavHl I guess your problem is much more general and your proposal will only solve it partially if at all.

From my experience using DI, it lacks notion of scopes for resources. Like "app", "request", "task", etc... I've solved the problem by doing scopes explicitly, i.e. injecting context manager that will initialize resources (e.g. db sessions) and storing them into contextvars. Not exactly great solution, but works when you have mix of different "apps" in your project (fastapi, typer, celery, alembic, etc...) but still need common setup code to run (configuring logging, apms, service discovery, etc...).

Introducing scopes would probably require significant re-engeneering of how DI itself works, so for the time being my advice for you would be to implement your own Lifespan the way you proposed. My implementation is intentionally dumb.

Also, take a look at ContextLocalSingleton. This is not a 100% replacement for resources, but might help in certain situations.

[Totorokrut]

Any chance to get it working with DynamicContainer?

I took an example from the documentation and changed the container type to dynamic.

from contextlib import asynccontextmanager
from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject
from dependency_injector.ext.starlette import Lifespan
from fastapi import FastAPI, Request, Depends, APIRouter


class Connection: ...


@asynccontextmanager
async def init_database():
    print("opening database connection")
    yield Connection()
    print("closing database connection")


router = APIRouter()


@router.get("/")
@inject
async def index(request: Request, db: Connection = Depends(Provide["db"])):
    # use the database connection here
    return "OK!"


class Container2(containers.DynamicContainer):
    __self__ = providers.Self()
    db = providers.Resource(init_database)
    lifespan = providers.Singleton(Lifespan, __self__)
    app = providers.Singleton(FastAPI, lifespan=lifespan)
    _include_router = providers.Resource(
        app.provided.include_router.call(),
        router,
    )


if __name__ == "__main__":
    import uvicorn

    container = Container2()
    app = container.app()
    uvicorn.run(app, host="localhost", port=8000)

And it fails now with:

INFO:     Started server process [878575]
INFO:     Waiting for application startup.
ERROR:    Traceback (most recent call last):
  File "/home/lglassner/Work/listener/.venv/lib/python3.12/site-packages/starlette/routing.py", line 692, in lifespan
    async with self.lifespan_context(app) as maybe_state:
               ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/lglassner/Work/listener/.venv/lib/python3.12/site-packages/dependency_injector/ext/starlette.py", line 52, in __aenter__
    result = self.container.init_resources(self.resource_type)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'init_resources'

ERROR:    Application startup failed. Exiting.

The issue is that there is no __self__ for a dynamic container, as I understand it. At the moment of creating Lifespan in __init__ there is just container=None.

Maybe we can somehow link to an existing container at run time? Something like that:

if __name__ == "__main__":
    import uvicorn

    container = Container2()
    container.__self__ = container
    app = container.app()
    uvicorn.run(app, host="localhost", port=8000)