Merge pull request #26 from AndrewSergienko/feature/litestar_integration

LItestar integration example

examples/litestar_app.py

@@ -0,0 +1,75 @@
+import logging
+from abc import abstractmethod
+from typing import Protocol, Annotated
+
+import uvicorn
+from litestar import Controller, get, Litestar
+
+from dishka import Provider, Scope, provide
+from dishka.integrations.base import Depends
+from dishka.integrations.litestar import (
+    inject,
+    setup_dishka,
+    startup_dishka,
+    make_dishka_container,
+    shutdown_dishka
+)
+
+
+# app core
+class DbGateway(Protocol):
+    @abstractmethod
+    def get(self) -> str:
+        raise NotImplementedError
+
+
+class FakeDbGateway(DbGateway):
+    def get(self) -> str:
+        return "Hello"
+
+
+class Interactor:
+    def __init__(self, db: DbGateway):
+        self.db = db
+
+    def __call__(self) -> str:
+        return self.db.get()
+
+
+# app dependency logic
+class AdaptersProvider(Provider):
+    @provide(scope=Scope.REQUEST)
+    def get_db(self) -> DbGateway:
+        return FakeDbGateway()
+
+
+class InteractorProvider(Provider):
+    i1 = provide(Interactor, scope=Scope.REQUEST)
+
+
+class MainController(Controller):
+    path = '/'
+
+    @get()
+    @inject
+    async def index(self, *, interactor: Annotated[Interactor, Depends()]) -> str:
+        result = interactor()
+        return result
+
+
+def create_app():
+    logging.basicConfig(
+        level=logging.WARNING,
+        format='%(asctime)s  %(process)-7s %(module)-20s %(message)s',
+    )
+    app = Litestar(
+        route_handlers=[MainController],
+        on_startup=[startup_dishka],
+        on_shutdown=[shutdown_dishka],
+        before_request=make_dishka_container
+    )
+    return setup_dishka(app, [InteractorProvider(), AdaptersProvider()])
+
+
+if __name__ == "__main__":
+    uvicorn.run(create_app(), host="0.0.0.0", port=8000)

src/dishka/integrations/litestar.py

@@ -0,0 +1,51 @@
+from inspect import Parameter
+from typing import Optional, Sequence, get_type_hints
+
+from litestar import Litestar, Request
+
+from dishka import Provider, make_async_container
+from dishka.integrations.base import wrap_injection
+
+
+def inject(func):
+    hints = get_type_hints(func)
+    request_param = next(
+        (name for name, hint in hints.items() if hint is Request),
+        None,
+    )
+    if request_param:
+        additional_params = []
+    else:
+        request_param = "request"
+        additional_params = [Parameter(
+            name=request_param,
+            annotation=Optional[Request],
+            kind=Parameter.KEYWORD_ONLY,
+        )]
+
+    return wrap_injection(
+        func=func,
+        remove_depends=True,
+        container_getter=lambda _, r: r[request_param].state.dishka_container,
+        additional_params=additional_params,
+        is_async=True,
+    )
+
+
+async def make_dishka_container(request: Request):
+    request_container = await request.app.state.dishka_container().__aenter__()
+    request.state.dishka_container = request_container
+
+
+async def startup_dishka(app: Litestar):
+    container = await app.state.dishka_container_wrapper.__aenter__()
+    app.state.dishka_container = container
+
+
+async def shutdown_dishka(app: Litestar):
+    await app.state.dishka_container_wrapper.__aexit__(None, None, None)
+
+
+def setup_dishka(app: Litestar, providers: Sequence[Provider]) -> Litestar:
+    app.state.dishka_container_wrapper = make_async_container(*providers)
+    return app