🪛 [Proposal] - Add new on_startup/on_shutdown events

starlette/events/context.py

@@ -0,0 +1,40 @@
+import typing
+from typing import TypeVar
+
+from starlette._utils import is_async_callable
+from starlette.types import Receive, Scope, Send
+
+_T = TypeVar("_T")
+
+
+class AyncLifespanContextManager:
+    """
+    Bridges the on_startup/on_shutdown to the new async context manager.
+    """
+
+    def __init__(
+        self,
+        on_startup: typing.Optional[typing.Sequence[typing.Callable]] = None,
+        on_shutdown: typing.Optional[typing.Sequence[typing.Callable]] = None,
+    ) -> None:
+        self.on_startup = [] if on_startup is None else list(on_startup)
+        self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)
+
+    def __call__(self: _T, app: object) -> _T:
+        return self
+
+    async def __aenter__(self) -> None:
+        for handler in self.on_startup:
+            if is_async_callable(handler):
+                await handler()
+            else:
+                handler()
+
+    async def __aexit__(
+        self, scope: Scope, receive: Receive, send: Send, **kwargs: typing.Any
+    ) -> None:
+        for handler in self.on_shutdown:
+            if is_async_callable(handler):
+                await handler()
+            else:
+                handler()

starlette/routing.py

@@ -13,6 +13,7 @@
 from starlette.concurrency import run_in_threadpool
 from starlette.convertors import CONVERTOR_TYPES, Convertor
 from starlette.datastructures import URL, Headers, URLPath
+from starlette.events.context import AyncLifespanContextManager
 from starlette.exceptions import HTTPException
 from starlette.middleware import Middleware
 from starlette.requests import Request
@@ -588,13 +589,20 @@ def __init__(
         self.on_startup = [] if on_startup is None else list(on_startup)
         self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)
 
+        assert lifespan is None or (
+            on_startup is None and on_shutdown is None
+        ), "Use either 'lifespan' or 'on_startup'/'on_shutdown', not both."
+
         if on_startup or on_shutdown:
             warnings.warn(
                 "The on_startup and on_shutdown parameters are deprecated, and they "
                 "will be removed on version 1.0. Use the lifespan parameter instead. "
                 "See more about it on https://www.starlette.io/lifespan/.",
                 DeprecationWarning,
             )
+            lifespan = self.handle_lifespan_events(
+                on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan
+            )
 
         if lifespan is None:
             self.lifespan_context: Lifespan = _DefaultLifespan(self)
@@ -620,6 +628,20 @@ def __init__(
         else:
             self.lifespan_context = lifespan
 
+    def handle_lifespan_events(
+        self,
+        on_startup: typing.Optional[typing.Sequence[typing.Callable]] = None,
+        on_shutdown: typing.Optional[typing.Sequence[typing.Callable]] = None,
+        lifespan: typing.Optional[Lifespan] = None,
+    ) -> typing.Any:
+        if on_startup or on_shutdown:
+            return AyncLifespanContextManager(
+                on_startup=on_startup, on_shutdown=on_shutdown
+            )
+        elif lifespan is not None:
+            return lifespan
+        return None
+
     async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
         if scope["type"] == "websocket":
             websocket_close = WebSocketClose()