RQ integration

docs/integrations/index.rst

@@ -21,6 +21,7 @@ You can create custom integrations for your framework of choice.
    flask
    grpcio
    litestar
+   rq
    sanic
    starlette
    taskiq
@@ -52,7 +53,7 @@ You can create custom integrations for your framework of choice.
 
    * - :ref:`Flask`
      -
-     -
+     - :ref:`RQ`
      -
    * -  :ref:`Litestar`
      -
@@ -112,4 +113,3 @@ For FastAPI it will look like:
    app = FastAPI()
    container = make_async_container(your_provider, FastapiProvider())
    setup_dishka(container, app)
-

docs/integrations/rq.rst

@@ -0,0 +1,50 @@
+.. _rq:
+
+rq
+===========================================
+
+Though it is not required, you can use dishka-rq integration. It features:
+
+* automatic REQUEST and SESSION scope management using Worker subclass
+* automatic injection of dependencies into job function.
+
+How to use
+****************
+
+1. Create provider and container as usual.
+
+.. code-block:: python
+
+  class StrProvider(Provider):
+      @provide(scope=Scope.REQUEST)
+      def hello(self) -> str:
+          return "Hello"
+
+  provider = StrProvider()
+  container = make_container(provider)
+
+2. Import.
+
+.. code-block:: python
+
+  from dishka import FromDishka
+
+3. Mark those of your job functions parameters which are to be injected with ``FromDishka[]``
+
+.. code-block:: python
+
+  def hello_world(hello: FromDishka[str]):
+      return f"{hello} world!"
+
+4. Run you worker using your container and DishkaWorker subclass.
+
+.. code-block:: python
+
+  conn = Redis()
+  queues = ["default"]
+  worker = DishkaWorker(container=container, queues=queues, connection=conn)
+  worker.work(with_scheduler=True)
+
+.. code-block:: shell
+
+  python run_worker.py

examples/integrations/rq/enqueue.py

@@ -0,0 +1,19 @@
+from time import sleep
+
+from redis import Redis
+from rq import Queue
+from tasks import hello_world
+
+if __name__ == "__main__":
+    connection = Redis()
+    queue = Queue(
+        name="default",
+        connection=connection,
+    )
+    job = queue.enqueue(hello_world)
+
+    res = job.result
+    while not res:
+        res = job.result
+        sleep(1)
+    print(res)

examples/integrations/rq/run_worker.py

@@ -0,0 +1,24 @@
+from redis import Redis
+
+from dishka import Provider, Scope, make_container, provide
+from dishka.integrations.rq import DishkaWorker
+
+
+class StrProvider(Provider):
+    @provide(scope=Scope.REQUEST)
+    def hello(self) -> str:
+        return "Hello"
+
+
+def setup_worker() -> DishkaWorker:
+    provider = StrProvider()
+    container = make_container(provider)
+    queues = ["default"]
+    conn = Redis()
+    worker = DishkaWorker(container=container, queues=queues, connection=conn)
+    return worker
+
+
+if __name__ == "__main__":
+    worker = setup_worker()
+    worker.work(with_scheduler=True)

examples/integrations/rq/tasks.py

@@ -0,0 +1,5 @@
+from dishka import FromDishka
+
+
+def hello_world(hello: FromDishka[str]):
+    return f"{hello} world!"

noxfile.py

@@ -50,6 +50,8 @@ def get_tests(self) -> str:
     IntegrationEnv("grpcio", "latest"),
     IntegrationEnv("litestar", "230"),
     IntegrationEnv("litestar", "latest"),
+    IntegrationEnv("rq", "200"),
+    IntegrationEnv("rq", "latest"),
     IntegrationEnv("sanic", "23121"),
     IntegrationEnv("sanic", "latest"),
     IntegrationEnv("starlette", "0270"),
@@ -88,4 +90,4 @@ def real_world(session: nox.Session) -> None:
         *INSTALL_CMD,
         "-r", "examples/real_world/requirements_test.txt",
     )
-    session.run(*CMD, "examples/real_world/tests/")
+    session.run(*CMD, "examples/real_world/tests/")

requirements/rq-200.txt

@@ -0,0 +1,3 @@
+-r test.txt
+rq==2.0.0
+fakeredis==2.26.2

requirements/rq-latest.txt

@@ -0,0 +1,3 @@
+-r test.txt
+rq==2.0.0
+fakeredis==2.26.2

src/dishka/integrations/rq.py

@@ -0,0 +1,77 @@
+from collections.abc import Callable
+from inspect import signature
+from typing import Any, get_type_hints
+
+from rq import Queue, Worker
+from rq.job import Job
+
+from dishka import Container
+from dishka.integrations.base import default_parse_dependency
+
+
+class DishkaWorker(Worker):
+    """Custom RQ Worker class with Dishka DI support."""
+
+    def __init__(
+        self,
+        *args,
+        container: Container,
+        **kwargs,
+    ) -> None:
+        """Sets up class and container."""
+        super().__init__(*args, **kwargs)
+        self.dishka_container = container
+
+    def perform_job(self, job: Job, queue: Queue) -> bool:
+        """Performs job call"""
+        request_container = self.dishka_container().__enter__()
+        self.inject_deps(job, request_container)
+        job_result = super().perform_job(job, queue)
+        request_container.close()
+        return job_result
+
+    def inject_deps(self, job: Job, container: Container) -> None:
+        """Injects dependencies into using the Dishka container.
+
+        Args:
+            job: The RQ job to inject dependencies into.
+        """
+        if job.func:
+            dependencies = self._build_dependencies(job.func)
+            updated_kwargs = self._build_kwargs(dependencies, container)
+            if isinstance(job.kwargs, dict):
+                job.kwargs.update(updated_kwargs)
+
+    def teardown(self) -> None:
+        """Closes DI container on worker shutdown."""
+        self.dishka_container.close()
+        super().teardown()
+
+    @classmethod
+    def _build_dependencies(
+        cls, callable_: Callable[..., Any],
+    ) -> dict[str, Any]:
+        """Builds dependencies for the given callable."""
+        dependencies = {}
+
+        for name, parameter in signature(callable_).parameters.items():
+            dep = default_parse_dependency(
+                parameter,
+                get_type_hints(callable_, include_extras=True).get(name, Any),
+            )
+            if dep is None:
+                continue
+            dependencies[name] = dep
+
+        return dependencies
+
+    def _build_kwargs(
+        self,
+        dependencies: dict,
+        request_container: Container,
+    ) -> dict[str, Any]:
+        """Buld kwargs dict for RQ job run."""
+        return {
+            name: request_container.get(dep.type_hint, component=dep.component)
+            for name, dep in dependencies.items()
+        }