Idea of LifespanManager to have multiple lifespans in app

[uriyyo]

The old approach with on_startup, on_shutdown is deprecated and will be removed in future versions of starlette.

The new lifespan approach has a small issue in that you can't have multiple lifespans defined. It's great to have a separate lifespan for db connection, cache connection, .etc.

I have created library called fastapi-lifespan-manager which aimed to fixed this issue.

I am happy to open PR to add this feature to fastapi cause the library is pretty small itself.

Usage example:

from typing import AsyncIterator

from fastapi import FastAPI
from pydantic import AnyUrl, BaseSettings
from redis.asyncio import Redis
from sqlalchemy.ext.asyncio import create_async_engine

from fastapi_lifespan_manager import LifespanManager, State


class Settings(BaseSettings):
    DB_URL: AnyUrl
    REDIS_DSN: AnyUrl | None = None


settings = Settings()

manager = LifespanManager()


# Lifespan functions to connect to the database
@manager.add
async def setup_db(app: FastAPI) -> AsyncIterator[State]:
    engine = await create_async_engine(settings.DB_URL)

    yield {"db": engine}

    await engine.dispose()


# Lifespan functions to connect to the cache, will be skipped if REDIS_DSN is None
if settings.REDIS_DSN:
    @manager.add
    async def setup_cache(app: FastAPI) -> AsyncIterator[State]:
        redis = await Redis.from_url(settings.REDIS_DSN)

        yield {"cache": redis}

        await redis.close()
        await redis.wait_closed()

app = FastAPI(lifespan=manager)

[ThirVondukr]

@uriyyo You could use this

import contextlib
from collections.abc import AsyncIterator, Callable, Sequence
from contextlib import AbstractAsyncContextManager

from fastapi import FastAPI


@contextlib.asynccontextmanager
async def _manager(
    app: FastAPI,
    lifespans: Sequence[Callable[[FastAPI], AbstractAsyncContextManager[None]]],
) -> AsyncIterator[None]:
    exit_stack = contextlib.AsyncExitStack()
    async with exit_stack:
        for lifespan in lifespans:
            await exit_stack.enter_async_context(lifespan(app))
        yield


class Lifespans:
    def __init__(
        self,
        lifespans: Sequence[Callable[[FastAPI], AbstractAsyncContextManager[None]]],
    ) -> None:
        self.lifespans = lifespans

    def __call__(self, app: FastAPI) -> AbstractAsyncContextManager[None]:
        self.app = app
        return _manager(app, lifespans=self.lifespans)


@contextlib.asynccontextmanager
async def one(app: FastAPI) -> AsyncIterator[None]:
    print("One")
    yield
    print("End One")


@contextlib.asynccontextmanager
async def two(app: FastAPI) -> AsyncIterator[None]:
    print("Two")
    yield
    print("End Two")


app = FastAPI(lifespan=Lifespans([one, two]))

[beornharris]

Thanks both for your help and inspiration! I wouldnt have found my solution without you...
I would like to suggest a (maybe) simpler approach though

from contextlib import asynccontextmanager, AsyncExitStack
from typing import List
from fastapi import FastAPI, APIRouter

def app_lifespan(lifespans: List):
    @asynccontextmanager
    async def _lifespan_manager(app: FastAPI):
        exit_stack = AsyncExitStack()
        async with exit_stack:
            for lifespan in lifespans:
                await exit_stack.enter_async_context(lifespan(app))
            yield
            
    return _lifespan_manager

@asynccontextmanager
async def one(app: FastAPI) -> AsyncIterator[None]:
    print("One")
    yield
    print("End One")


@asynccontextmanager
async def two(app: FastAPI) -> AsyncIterator[None]:
    print("Two")
    yield
    print("End Two")


app = FastAPI(lifespan=app_lifespan([one, two]))

I confess I did not run this example, but the code is ripped from a working version that I am using....so apologies for any oversights.
I hope this is useful/helpful to someone.