update modern-di

Author: lesnik512

app/application.py

@@ -1,6 +1,3 @@
-import contextlib
-import typing
-
 import fastapi
 import modern_di_fastapi
 from advanced_alchemy.exceptions import DuplicateKeyError
@@ -15,40 +12,31 @@ def include_routers(app: fastapi.FastAPI) -> None:
     app.include_router(ROUTER, prefix="/api")
 
 
-class AppBuilder:
-    def __init__(self) -> None:
-        self.app: fastapi.FastAPI = fastapi.FastAPI(
-            lifespan=self.lifespan_manager,
-        )
-        self.bootstrapper = FastAPIBootstrapper(
-            bootstrap_config=FastAPIConfig(
-                application=self.app,
-                service_name=settings.service_name,
-                service_version=settings.service_version,
-                service_environment=settings.service_environment,
-                service_debug=settings.service_debug,
-                opentelemetry_endpoint=settings.opentelemetry_endpoint,
-                sentry_dsn=settings.sentry_dsn,
-                cors_allowed_origins=settings.cors_allowed_origins,
-                cors_allowed_methods=settings.cors_allowed_methods,
-                cors_allowed_headers=settings.cors_allowed_headers,
-                cors_exposed_headers=settings.cors_exposed_headers,
-                logging_buffer_capacity=settings.logging_buffer_capacity,
-                swagger_offline_docs=settings.swagger_offline_docs,
-            ),
-        )
-        self.bootstrapper.bootstrap()
-        self.di_container = modern_di_fastapi.setup_di(self.app)
-        include_routers(self.app)
-        self.app.add_exception_handler(
-            DuplicateKeyError,
-            exceptions.duplicate_key_error_handler,  # type: ignore[arg-type]
-        )
-
-    @contextlib.asynccontextmanager
-    async def lifespan_manager(self, _: fastapi.FastAPI) -> typing.AsyncIterator[dict[str, typing.Any]]:
-        async with self.di_container:
-            yield {}
-
-
-application = AppBuilder().app
+def build_app() -> fastapi.FastAPI:
+    bootstrapper = FastAPIBootstrapper(
+        bootstrap_config=FastAPIConfig(
+            service_name=settings.service_name,
+            service_version=settings.service_version,
+            service_environment=settings.service_environment,
+            service_debug=settings.service_debug,
+            opentelemetry_endpoint=settings.opentelemetry_endpoint,
+            sentry_dsn=settings.sentry_dsn,
+            cors_allowed_origins=settings.cors_allowed_origins,
+            cors_allowed_methods=settings.cors_allowed_methods,
+            cors_allowed_headers=settings.cors_allowed_headers,
+            cors_exposed_headers=settings.cors_exposed_headers,
+            logging_buffer_capacity=settings.logging_buffer_capacity,
+            swagger_offline_docs=settings.swagger_offline_docs,
+        ),
+    )
+    app: fastapi.FastAPI = bootstrapper.bootstrap()
+    modern_di_fastapi.setup_di(app)
+    include_routers(app)
+    app.add_exception_handler(
+        DuplicateKeyError,
+        exceptions.duplicate_key_error_handler,  # type: ignore[arg-type]
+    )
+    return app
+
+
+application = build_app()

pyproject.toml

@@ -30,6 +30,7 @@ dev = [
     "pytest",
     "pytest-cov",
     "pytest-asyncio",
+    "asgi_lifespan",
     "ruff",
     "mypy",
     "asyncpg-stubs",

tests/conftest.py

@@ -1,29 +1,36 @@
 import typing
 
+import fastapi
 import modern_di
 import modern_di_fastapi
 import pytest
+from asgi_lifespan import LifespanManager
 from httpx import ASGITransport, AsyncClient
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app import ioc
-from app.application import application
+from app.application import build_app
 
 
 @pytest.fixture
-async def client() -> typing.AsyncIterator[AsyncClient]:
+async def app() -> typing.AsyncIterator[fastapi.FastAPI]:
+    app_ = build_app()
+    async with LifespanManager(app_):
+        yield app_
+
+
+@pytest.fixture
+async def client(app: fastapi.FastAPI) -> typing.AsyncIterator[AsyncClient]:
     async with AsyncClient(
-        transport=ASGITransport(app=application),
+        transport=ASGITransport(app=app),
         base_url="http://test",
     ) as client:
         yield client
 
 
 @pytest.fixture
-async def di_container() -> typing.AsyncIterator[modern_di.Container]:
-    di_container_: typing.Final = modern_di_fastapi.fetch_di_container(application)
-    async with di_container_:
-        yield di_container_
+def di_container(app: fastapi.FastAPI) -> modern_di.Container:
+    return modern_di_fastapi.fetch_di_container(app)
 
 
 @pytest.fixture(autouse=True)

tests/test_main.py

@@ -1,24 +1,17 @@
 import runpy
 from unittest import mock
 
-import fastapi
 import modern_di
 import pytest
 
 from app import ioc
-from app.application import AppBuilder
 
 
 def test_main(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr("granian.Granian", mock.Mock())
     runpy.run_module("app.__main__", run_name="__main__")
 
 
-async def test_app_lifespan() -> None:
-    async with AppBuilder().lifespan_manager(fastapi.FastAPI()):
-        pass
-
-
 async def test_session() -> None:
     async with (
         modern_di.Container(scope=modern_di.Scope.APP) as container,