Add Quart integration

docs/integrations/index.rst

@@ -25,6 +25,7 @@ You can create custom integrations for your framework of choice.
    starlette
    taskiq
    telebot
+   quart
    adding_new
    celery
 
@@ -67,6 +68,10 @@ You can create custom integrations for your framework of choice.
      -
      -
      -
+   * -  :ref:`Quart`
+     -
+     -
+     -
 
 If you have another framework  refer :ref:`adding_new`
 

docs/integrations/quart.rst

@@ -0,0 +1,94 @@
+.. _quart:
+
+Quart
+===========================================
+
+Though it is not required, you can use dishka-quart integration. It features:
+
+* automatic REQUEST scope management using before/after request hooks
+* passing ``Request`` and ``Websocket`` objects as context data to providers
+* injection of dependencies into route and WebSocket handlers using decorator
+* optional auto-injection for all handlers
+
+
+How to use
+****************
+
+1. Import
+
+..  code-block:: python
+
+    from dishka.integrations.quart import (
+        FromDishka,
+        QuartProvider,
+        inject,
+        setup_dishka,
+    )
+    from dishka import make_async_container, Provider, provide, Scope
+
+2. Create provider. You can use ``quart.Request`` or ``quart.Websocket`` as factory parameters to access HTTP or WebSocket request.
+They are available on ``REQUEST`` and ``SESSION`` scopes respectively.
+
+.. code-block:: python
+
+    class YourProvider(Provider):
+        @provide(scope=Scope.REQUEST)
+        def create_x(self, request: Request) -> X:
+             ...
+
+3. Mark those of your handlers parameters which are to be injected with ``FromDishka[]``
+
+.. code-block:: python
+
+    @app.route('/')
+    async def endpoint(
+        interactor: FromDishka[Interactor],
+    ) -> str:
+        result = await interactor()
+        return result
+
+3a. *(optional)* decorate them using ``@inject`` if not using auto-injection
+
+.. code-block:: python
+
+    @app.route('/')
+    @inject
+    async def endpoint(
+        interactor: FromDishka[Interactor],
+    ) -> str:
+        result = await interactor()
+        return result
+
+4. *(optional)* Use ``QuartProvider()`` when creating container if you need ``Request`` or ``Websocket`` in providers
+
+.. code-block:: python
+
+    container = make_async_container(YourProvider(), QuartProvider())
+
+5. Setup dishka integration. ``auto_inject=True`` is required unless you explicitly use ``@inject`` decorator. It is important here to call it after registering all views and blueprints.
+
+.. code-block:: python
+
+    setup_dishka(container=container, app=app, auto_inject=True)
+
+
+WebSockets
+**********************
+
+.. include:: _websockets.rst
+
+For WebSocket handlers, you can use the same pattern as with regular routes. The WebSocket connection is available in ``SESSION`` scope:
+
+.. code-block:: python
+
+    @app.websocket("/")
+    @inject
+    async def websocket_handler(
+        ws: FromDishka[Websocket],
+        interactor: FromDishka[Interactor],
+    ) -> None:
+        await ws.accept()
+        while True:
+            data = await ws.receive()
+            result = await interactor.process(data)
+            await ws.send(result)

examples/integrations/quart_app.py

@@ -0,0 +1,78 @@
+from abc import abstractmethod
+from typing import Protocol
+
+from quart import Quart
+
+from dishka import (
+    Provider,
+    Scope,
+    make_async_container,
+    provide,
+)
+from dishka.integrations.quart import (
+    QuartProvider,
+    FromDishka,
+    inject,
+    setup_dishka,
+)
+
+
+# app core
+class DbGateway(Protocol):
+    @abstractmethod
+    async def get(self) -> str:
+        raise NotImplementedError
+
+
+class FakeDbGateway(DbGateway):
+    async def get(self) -> str:
+        return "Hello"
+
+
+class Interactor:
+    def __init__(self, db: DbGateway):
+        self.db = db
+
+    async def __call__(self) -> str:
+        return await 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)
+
+
+# presentation layer
+app = Quart(__name__)
+
+
+@app.get("/manual")
+@inject
+async def manual_inject(
+        *,
+        interactor: FromDishka[Interactor],
+) -> str:
+    result = await interactor()
+    return result
+
+
+@app.get("/auto")
+async def auto_inject(
+        *,
+        interactor: FromDishka[Interactor],
+) -> str:
+    result = await interactor()
+    return result
+
+
+container = make_async_container(AdaptersProvider(), InteractorProvider(), QuartProvider())
+setup_dishka(container=container, app=app, auto_inject=True)
+
+if __name__ == "__main__":
+    app.run()

noxfile.py

@@ -60,6 +60,8 @@ def get_tests(self) -> str:
     IntegrationEnv("telebot", "latest"),
     IntegrationEnv("celery", "540"),
     IntegrationEnv("celery", "latest"),
+    IntegrationEnv("quart", "0181"),
+    IntegrationEnv("quart", "latest"),
 ]
 
 

requirements/quart-0181.txt

@@ -0,0 +1,2 @@
+-r test.txt
+quart==0.18.1

requirements/quart-latest.txt

@@ -0,0 +1,2 @@
+-r test.txt
+quart

src/dishka/integrations/quart.py

@@ -0,0 +1,92 @@
+__all__ = [
+    "FromDishka",
+    "QuartProvider",
+    "inject",
+    "setup_dishka",
+]
+
+from collections.abc import Callable
+from functools import wraps
+from typing import Any, ParamSpec, TypeAlias, TypeVar, cast
+
+from quart import Blueprint, Quart, Request, Websocket, g, request, websocket
+from quart.typing import RouteCallable, WebsocketCallable
+
+from dishka import AsyncContainer, FromDishka, Provider, Scope, from_context
+from dishka.integrations.base import is_dishka_injected, wrap_injection
+
+P = ParamSpec("P")
+T = TypeVar("T")
+Scaffold: TypeAlias = Quart | Blueprint
+
+
+def inject(func: Callable[P, T]) -> Callable[P, T]:
+    return wrap_injection(
+        func=func,
+        is_async=True,
+        container_getter=lambda _, p: g.dishka_container,
+    )
+
+
+class QuartProvider(Provider):
+    request = from_context(Request, scope=Scope.REQUEST)
+    websocket = from_context(Websocket, scope=Scope.SESSION)
+
+
+class ContainerMiddleware:
+    def __init__(self, container: AsyncContainer) -> None:
+        self.container = container
+
+    async def enter_request(self) -> None:
+        wrapper = self.container({Request: request}, scope=Scope.REQUEST)
+        g.dishka_container_wrapper = wrapper
+        g.dishka_container = await wrapper.__aenter__()
+
+    async def enter_websocket(self) -> None:
+        wrapper = self.container({Websocket: websocket}, scope=Scope.SESSION)
+        g.dishka_container_wrapper = wrapper
+        g.dishka_container = await wrapper.__aenter__()
+
+    async def exit_scope(self, *_args: Any, **_kwargs: Any) -> None:
+        if hasattr(g, "dishka_container"):
+            await g.dishka_container.close()
+
+
+def _make_wrapper(func: Any) -> Any:
+    @wraps(func)
+    async def wrapped(*args: Any, **kwargs: Any) -> Any:
+        injected = inject(func)
+        result = injected(*args, **kwargs)
+        if hasattr(result, "__await__"):
+            result = await result
+        return result
+
+    return wrapped
+
+
+def _inject_routes(app: Scaffold) -> None:
+    for endpoint, func in app.view_functions.items():
+        if not is_dishka_injected(func):
+            wrapped = _make_wrapper(func)
+            if getattr(func, "websocket", False):
+                app.view_functions[endpoint] = cast(WebsocketCallable, wrapped)
+            else:
+                app.view_functions[endpoint] = cast(RouteCallable, wrapped)
+
+
+def setup_dishka(
+    container: AsyncContainer,
+    app: Quart,
+    *,
+    auto_inject: bool = False,
+) -> None:
+    middleware = ContainerMiddleware(container)
+    app.before_request(middleware.enter_request)
+    app.before_websocket(middleware.enter_websocket)
+    app.teardown_request(middleware.exit_scope)
+    app.teardown_websocket(middleware.exit_scope)
+
+    if auto_inject:
+        _inject_routes(app)
+        for blueprint in app.blueprints.values():
+            _inject_routes(blueprint)

tests/integrations/quart/__init__.py

@@ -0,0 +1,3 @@
+import pytest
+
+pytest.importorskip("quart")

tests/integrations/quart/test_quart.py

@@ -0,0 +1,85 @@
+from contextlib import asynccontextmanager
+from unittest.mock import Mock
+
+import pytest
+from quart import Quart
+
+from dishka import make_async_container
+from dishka.integrations.quart import FromDishka, inject, setup_dishka
+from ..common import (
+    APP_DEP_VALUE,
+    REQUEST_DEP_VALUE,
+    AppDep,
+    AppProvider,
+    RequestDep,
+)
+
+
+@asynccontextmanager
+async def dishka_app(view, provider):
+    app = Quart(__name__)
+    app.get("/")(inject(view))
+    container = make_async_container(provider)
+    setup_dishka(container=container, app=app)
+    yield app
+    await container.close()
+
+
+@asynccontextmanager
+async def dishka_auto_app(view, provider):
+    app = Quart(__name__)
+    container = make_async_container(provider)
+    setup_dishka(container=container, app=app, auto_inject=True)
+    app.route("/")(inject(view))
+    yield app
+    await container.close()
+
+
+async def handle_with_app(
+        a: FromDishka[AppDep],
+        mock: FromDishka[Mock],
+) -> None:
+    mock(a)
+
+
+@pytest.mark.parametrize("app_factory", [
+    dishka_app, dishka_auto_app,
+])
+@pytest.mark.asyncio
+async def test_app_dependency(app_provider: AppProvider, app_factory):
+    async with app_factory(handle_with_app, app_provider) as app:
+        test_client = app.test_client()
+        await test_client.get("/")
+        app_provider.mock.assert_called_with(APP_DEP_VALUE)
+        app_provider.app_released.assert_not_called()
+    app_provider.app_released.assert_called()
+
+
+async def handle_with_request(
+        a: FromDishka[RequestDep],
+        mock: FromDishka[Mock],
+) -> None:
+    mock(a)
+
+
+@pytest.mark.asyncio
+async def test_request_dependency(app_provider: AppProvider):
+    async with dishka_app(handle_with_request, app_provider) as app:
+        test_client = app.test_client()
+        await test_client.get("/")
+        app_provider.mock.assert_called_with(REQUEST_DEP_VALUE)
+        app_provider.request_released.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_request_dependency2(app_provider: AppProvider):
+    async with dishka_app(handle_with_request, app_provider) as app:
+        test_client = app.test_client()
+        await test_client.get("/")
+        app_provider.mock.assert_called_with(REQUEST_DEP_VALUE)
+        app_provider.request_released.assert_called_once()
+        app_provider.mock.reset_mock()
+        app_provider.request_released.reset_mock()
+        await test_client.get("/")
+        app_provider.mock.assert_called_with(REQUEST_DEP_VALUE)
+        app_provider.request_released.assert_called_once()