Skip to content

Commit

Permalink
Add Starlette lifespan handler implementation
Browse files Browse the repository at this point in the history
  • Loading branch information
ZipFile committed Aug 12, 2024
1 parent 2c998b8 commit c1c65b1
Show file tree
Hide file tree
Showing 5 changed files with 188 additions and 0 deletions.
39 changes: 39 additions & 0 deletions examples/miniapps/starlette-lifespan/README.rst
Original file line number Diff line number Diff line change
@@ -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
uvicorn --factory example:container.app
# or
python example.py
After that visit http://127.0.0.1:8000/ in your browser or use CLI command (``curl``, ``httpie``,
etc).
52 changes: 52 additions & 0 deletions examples/miniapps/starlette-lifespan/example.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
#!/usr/bin/env python
# uvicorn --factory example:container.app

from logging import basicConfig

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


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",
)
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,
)
3 changes: 3 additions & 0 deletions examples/miniapps/starlette-lifespan/requirements.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
dependency-injector
starlette
uvicorn
53 changes: 53 additions & 0 deletions src/dependency_injector/ext/starlette.py
Original file line number Diff line number Diff line change
@@ -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
41 changes: 41 additions & 0 deletions tests/unit/ext/test_starlette.py
Original file line number Diff line number Diff line change
@@ -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

0 comments on commit c1c65b1

Please sign in to comment.