feat: hiredis optional; add option to run in existing process

feat: `hiredis` optional; add option to run in existing process

Author: cofin

.pre-commit-config.yaml

@@ -28,7 +28,7 @@ repos:
       - id: codespell
         exclude: "pdm.lock"
   - repo: https://github.com/psf/black
-    rev: 23.9.1
+    rev: 23.10.0
     hooks:
       - id: black
         args: [--config=./pyproject.toml]

litestar_saq/base.py

@@ -13,10 +13,12 @@
 
 if TYPE_CHECKING:
     from collections.abc import Collection
-    from signal import Signals
 
     from redis.asyncio.client import Redis
-    from saq.types import DumpType, Function, LoadType, PartialTimersDict, ReceivesContext
+    from saq.types import DumpType as SaqDumpType
+    from saq.types import Function, LoadType, PartialTimersDict, ReceivesContext
+
+    from litestar_saq.config import DumpType
 
 
 @dataclass
@@ -64,7 +66,7 @@ def __init__(
         queue_namespace: str | None = None,
     ) -> None:
         self._namespace = queue_namespace if queue_namespace is not None else "saq"
-        super().__init__(redis, name, dump, load, max_concurrent_ops)
+        super().__init__(redis, name, cast("SaqDumpType", dump), load, max_concurrent_ops)
 
     def namespace(self, key: str) -> str:
         """Make the namespace unique per app."""
@@ -85,8 +87,10 @@ def job_id(self, job_key: str) -> str:
 class Worker(SaqWorker):
     """Worker."""
 
+    """
     # same issue: https://github.com/samuelcolvin/arq/issues/182
     SIGNALS: list[Signals] = []
+    """
 
     def __init__(
         self,
@@ -101,7 +105,9 @@ def __init__(
         after_process: ReceivesContext | None = None,
         timers: PartialTimersDict | None = None,
         dequeue_timeout: float = 0,
+        separate_process: bool = True,
     ) -> None:
+        self.separate_process = separate_process
         super().__init__(
             cast("SaqQueue", queue),
             functions,
@@ -117,5 +123,13 @@ def __init__(
 
     async def on_app_startup(self) -> None:
         """Attach the worker to the running event loop."""
-        loop = asyncio.get_running_loop()
-        _ = loop.create_task(self.start())
+        if not self.separate_process:
+            self.SIGNALS = []
+            loop = asyncio.get_running_loop()
+            _ = loop.create_task(self.start())
+
+    async def on_app_shutdown(self) -> None:
+        """Attach the worker to the running event loop."""
+        if not self.separate_process:
+            loop = asyncio.get_running_loop()
+            _ = loop.create_task(self.stop())

litestar_saq/cli.py

@@ -137,10 +137,11 @@ def run_worker_process(workers: list[Worker], logging_config: BaseLoggingConfig
         logging_config.configure()
     try:
         for i, worker_instance in enumerate(workers):
-            if i < len(workers) - 1:
-                loop.create_task(worker_instance.start())
-            else:
-                loop.run_until_complete(worker_instance.start())
+            if worker_instance.separate_process:
+                if i < len(workers) - 1:
+                    loop.create_task(worker_instance.start())
+                else:
+                    loop.run_until_complete(worker_instance.start())
     except KeyboardInterrupt:
         for worker in workers:
             loop.run_until_complete(worker.stop())

litestar_saq/config.py

@@ -2,13 +2,14 @@
 
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import TYPE_CHECKING, Collection, Mapping, TypeVar, cast
+from typing import TYPE_CHECKING, Callable, Collection, Dict, Mapping, TypeVar, cast
 
 from litestar.exceptions import ImproperlyConfiguredException
 from litestar.serialization import decode_json, encode_json
 from redis.asyncio import ConnectionPool, Redis
 from saq.queue import Queue as SaqQueue
-from saq.types import DumpType, LoadType, PartialTimersDict, QueueInfo, QueueStats, ReceivesContext
+from saq.types import DumpType as SaqDumpType
+from saq.types import LoadType, PartialTimersDict, QueueInfo, QueueStats, ReceivesContext
 
 from litestar_saq._util import import_string, module_to_os_path
 from litestar_saq.base import CronJob, Job, Queue, Worker
@@ -22,6 +23,7 @@
 
 T = TypeVar("T")
 TaskQueue = Queue | SaqQueue
+DumpType = SaqDumpType | Callable[[Dict], bytes]
 
 
 def serializer(value: Any) -> str:
@@ -112,16 +114,6 @@ def signature_namespace(self) -> dict[str, Any]:
             "TaskQueues": TaskQueues,
         }
 
-    async def on_shutdown(self, app: Litestar) -> None:
-        """Disposes of the SAQ Workers.
-
-        Args:
-            app: The ``Litestar`` instance.
-
-        Returns:
-            None
-        """
-
     def provide_queues(self, state: State) -> TaskQueues:
         """Provide the configured job queues.
 
@@ -161,7 +153,7 @@ def get_queues(self) -> TaskQueues:
                 queue_namespace=self.namespace,
                 redis=self.get_redis(),
                 name=queue_config.name,
-                dump=self.json_serializer,
+                dump=cast("SaqDumpType", self.json_serializer),
                 load=self.json_deserializer,
                 max_concurrent_ops=queue_config.max_concurrent_ops,
             )

litestar_saq/plugin.py

@@ -66,9 +66,12 @@ def on_app_init(self, app_config: AppConfig) -> AppConfig:
             )
             app_config.route_handlers.append(build_controller(self._config.web_path))
         app_config.on_startup.append(self._config.update_app_state)
-        app_config.on_shutdown.append(self._config.on_shutdown)
         app_config.signature_namespace.update(self._config.signature_namespace)
-
+        workers = self.get_workers()
+        for worker in workers:
+            if not worker.separate_process:
+                app_config.on_startup.append(worker.on_app_startup)
+                app_config.on_shutdown.append(worker.on_app_shutdown)
         return app_config
 
     def get_workers(self) -> list[Worker]:
@@ -88,6 +91,7 @@ def get_workers(self) -> list[Worker]:
                 after_process=queue_config.after_process,
                 timers=queue_config.timers,
                 dequeue_timeout=queue_config.dequeue_timeout,
+                separate_process=queue_config.separate_process,
             )
             for queue_config in self._config.queue_configs
         )