Support starlette "lifespan" context for application

[uSpike]

First check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.
• After submitting this, I commit to:
  • Read open issues with questions until I find 2 issues where I can help someone and add a comment to help there.
  • Or, I already hit the "watch" button in this repository to receive notifications and I commit to help at least 2 people that ask questions in the future.
  • Implement a Pull Request for a confirmed bug.

Description

Starlette added support for the lifespan context function in 0.13.5. It's not well documented, but the code does suggest:

        # The lifespan context function is a newer style that replaces
        # on_startup / on_shutdown handlers. Use one or the other, not both.

For my purposes, it's much more convenient to use a (async)contextmanager for the startup and shutdown events of my application. It would be nice to have the option

The solution you would like

I'd like an easier way, and one that's officially supported, to use the lifespan context function.

from fastapi import FastAPI

async def lifespan(app):
    print("startup")
    async with SomeResource():
        yield
    print("shutdown")

app = FastAPI(lifespan=lifespan)

or

from fastapi import FastAPI

app = FastAPI()

@app.lifespan
async def lifespan(app):
    print("startup")
    async with SomeResource():
        yield
    print("shutdown")

Describe alternatives you've considered

I can already accomplish this simply by doing:

from fastapi import FastAPI

async def lifespan(app):
    print("startup")
    async with SomeResource():
        yield
    print("shutdown")

app = FastAPI()
app.router.lifespan_context = lifespan

however this is not officially supported and would likely break if accidentally using app.on_event in addition.

One could also do nasty stuff with __aenter__ and __aexit__:

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup()
    print("startup")
    app.state.resource = SomeResource()
    await app.state.resource.__aenter__()

@app.on_event("shutdown")
async def startup()
    print("shutdown")
    await app.state.resource.__aexit__(None, None, None)

but that seems quite ugly to me.

Environment

• OS: Linux
• FastAPI Version: 0.63.0
• Python version: 3.8.5

[juntatalor]

@uSpike Hi! Did you look into asgi-lifespan package?

[uSpike]

@juntatalor I did not look in to asgi-lifespan. It seems strange to use something outside of starlette, when starlette already supports the lifespan context:

from starlette import Starlette

async def lifespan(app):
    async with SomeResource():
        yield

app = Starlette(lifespan=lifespan)

What I'm looking for in this request is to have the starlette functionality exposed in fastapi.

[sm-Fifteen]

See also issue #617 (which is about potential ways to use lifespan with FastAPI's dependency injection system and potential APIs for that) and encode/starlette#799 (which implemented the lifespan context manager).

[ariloulaleelay]

I use such solution:

from copy import deepcopy

from fastapi import Depends, Request, WebSocket
from merge_arge import merge_args
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import Receive, Scope, Send

                           
class AppLifetimeDependencyMiddleware(BaseHTTPMiddleware):

    def __init__(self, **kwargs):
        kwargs = dict(kwargs)
        new_kwargs = {}
        for key in ('app', 'dispatch'):
            if key in kwargs:
                new_kwargs[key] = kwargs.pop(key)

        self.__storage = {}
        self.__kwargs = kwargs or {}
        super().__init__(**new_kwargs)

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        scope['app_lifetime_dependency'] = self.__storage
        scope['app_lifetime_dependency_kwargs'] = deepcopy(self.__kwargs)

        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        request = Request(scope, receive=receive)
        response = await self.dispatch_func(request, self.call_next)
        await response(scope, receive, send)

    async def dispatch(self, request, call_next):
        return await call_next(request)


def app_lifetime_dependency(f):
    key = id(f)

    @merge_args(f)
    async def wrapped(*args, request: Request = None, websocket: WebSocket = None, **kwargs):
        req = request
        if request is None:
            req = websocket
        if req is None:
            return f(*args, **kwargs)  # if none of request or websocket, then it's a direct call
        if key not in req.scope['app_lifetime_dependency']:
            result = await f(*args, **kwargs)
            # TODO add check for async_generator and proper handling
            req.scope['app_lifetime_dependency'][key] = result
        return req.scope['app_lifetime_dependency'][key]

    return wrapped


def get_config(request: Request):
    return request.scope['app_lifetime_dependency_kwargs']['config']


@app_lifetime_dependency
def get_db_pool(config = Depends(get_config)):
    conn = init_db_connection(config)
    return conn.pool()
    
def main():
    ...
    config = load_config_from_file()
    app.add_middleware(
        AppLifetimeDependencyMiddleware,
        config=config
    )
    

[tiangolo]

Thank you @uSpike! And thanks for the PR implementing it. 🚀 🍰

This will be available in FastAPI 0.93.0, released in the next few hours. 🎉

Sorry for the long delay! 🙈 I wanted to personally address each issue/PR and they piled up through time, but now I'm checking each one in order.

[Soberia]

Not having the app = FastAPI() line at the top of the module is such a pain!
lifespan also needs to be provided as a decorator.

@app.lifespan
async def lifespan(app: FastAPI) -> None:
    ...