Add Starlette lifespan handler implementation

Author: ZipFile

examples/miniapps/starlette-lifespan/README.rst

@@ -0,0 +1,39 @@
+Integration With Starlette-based Frameworks
+===========================================
+
+This is a `Starlette <https://www.starlette.io/>`_ +
+`Dependency Injector <https://python-dependency-injector.ets-labs.org/>`_ example application
+utilizing `lifespan API <https://www.starlette.io/lifespan/>`_.
+
+.. note::
+
+    Pretty much `any framework built on top of Starlette <https://www.starlette.io/third-party-packages/#frameworks>`_
+    supports this feature (`FastAPI <https://fastapi.tiangolo.com/advanced/events/#lifespan>`_,
+    `Xpresso <https://xpresso-api.dev/latest/tutorial/lifespan/>`_, etc...).
+
+Run
+---
+
+Create virtual environment:
+
+.. code-block:: bash
+
+    python -m venv env
+    . env/bin/activate
+
+Install requirements:
+
+.. code-block:: bash
+
+    pip install -r requirements.txt
+
+To run the application do:
+
+.. code-block:: bash
+
+    python example.py
+    # or (logging won't be configured):
+    uvicorn --factory example:container.app
+
+After that visit http://127.0.0.1:8000/ in your browser or use CLI command (``curl``, ``httpie``,
+etc).

examples/miniapps/starlette-lifespan/example.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python
+
+from logging import basicConfig, getLogger
+
+from dependency_injector.containers import DeclarativeContainer
+from dependency_injector.ext.starlette import Lifespan
+from dependency_injector.providers import Factory, Resource, Self, Singleton
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.routing import Route
+
+count = 0
+
+
+def init():
+    log = getLogger(__name__)
+    log.info("Inittializing resources")
+    yield
+    log.info("Cleaning up resources")
+
+
+async def homepage(request: Request) -> JSONResponse:
+    global count
+    response = JSONResponse({"hello": "world", "count": count})
+    count += 1
+    return response
+
+
+class Container(DeclarativeContainer):
+    __self__ = Self()
+    lifespan = Singleton(Lifespan, __self__)
+    logging = Resource(
+        basicConfig,
+        level="DEBUG",
+        datefmt="%Y-%m-%d %H:%M",
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    )
+    init = Resource(init)
+    app = Factory(
+        Starlette,
+        debug=True,
+        lifespan=lifespan,
+        routes=[Route("/", homepage)],
+    )
+
+
+container = Container()
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(
+        container.app,
+        factory=True,
+        # NOTE: `None` prevents uvicorn from configuring logging, which is
+        #       impossible via CLI
+        log_config=None,
+    )

examples/miniapps/starlette-lifespan/requirements.txt

@@ -0,0 +1,3 @@
+dependency-injector
+starlette
+uvicorn

src/dependency_injector/ext/starlette.py

@@ -0,0 +1,53 @@
+import sys
+from abc import ABCMeta, abstractmethod
+from typing import Any, Callable, Coroutine, Optional
+
+if sys.version_info >= (3, 11):  # pragma: no cover
+    from typing import Self
+else:  # pragma: no cover
+    from typing_extensions import Self
+
+from dependency_injector.containers import Container
+
+
+class Lifespan:
+    """A starlette lifespan handler performing container resource initialization and shutdown.
+
+    See https://www.starlette.io/lifespan/ for details.
+
+    Usage:
+
+    .. code-block:: python
+
+        from dependency_injector.containers import DeclarativeContainer
+        from dependency_injector.ext.starlette import Lifespan
+        from dependency_injector.providers import Factory, Self, Singleton
+        from starlette.applications import Starlette
+
+        class Container(DeclarativeContainer):
+            __self__ = Self()
+            lifespan = Singleton(Lifespan, __self__)
+            app = Factory(Starlette, lifespan=lifespan)
+
+    :param container: container instance
+    """
+
+    container: Container
+
+    def __init__(self, container: Container) -> None:
+        self.container = container
+
+    def __call__(self, app: Any) -> Self:
+        return self
+
+    async def __aenter__(self) -> None:
+        result = self.container.init_resources()
+
+        if result is not None:
+            await result
+
+    async def __aexit__(self, *exc_info: Any) -> None:
+        result = self.container.shutdown_resources()
+
+        if result is not None:
+            await result

tests/unit/ext/test_starlette.py

@@ -0,0 +1,41 @@
+from typing import AsyncIterator, Iterator
+from unittest.mock import ANY
+
+from pytest import mark
+
+from dependency_injector.containers import DeclarativeContainer
+from dependency_injector.ext.starlette import Lifespan
+from dependency_injector.providers import Resource
+
+
+class TestLifespan:
+    @mark.parametrize("sync", [False, True])
+    @mark.asyncio
+    async def test_context_manager(self, sync: bool) -> None:
+        init, shutdown = False, False
+
+        def sync_resource() -> Iterator[None]:
+            nonlocal init, shutdown
+
+            init = True
+            yield
+            shutdown = True
+
+        async def async_resource() -> AsyncIterator[None]:
+            nonlocal init, shutdown
+
+            init = True
+            yield
+            shutdown = True
+
+        class Container(DeclarativeContainer):
+            x = Resource(sync_resource if sync else async_resource)
+
+        container = Container()
+        lifespan = Lifespan(container)
+
+        async with lifespan(ANY) as scope:
+            assert scope is None
+            assert init
+
+        assert shutdown