DM-45281: Add function to build arq RedisSettings

changelog.d/20240718_082708_rra_DM_45281_queue.md

@@ -0,0 +1,3 @@
+### New features
+
+- Add new utility function `safir.arq.build_arq_redis_settings`, which constructs the `RedisSettings` object used to create an arq Redis queue from a Pydantic Redis DSN.

docs/user-guide/arq.rst

@@ -52,7 +52,7 @@ If your app uses a configuration system like ``pydantic.BaseSettings``, this exa
     from arq.connections import RedisSettings
     from pydantic import Field
     from pydantic_settings import BaseSettings
-    from safir.arq import ArqMode
+    from safir.arq import ArqMode, build_arq_redis_settings
     from safir.pydantic import EnvRedisDsn
 
 
@@ -61,22 +61,20 @@ If your app uses a configuration system like ``pydantic.BaseSettings``, this exa
             "redis://localhost:6379/1", validation_alias="APP_ARQ_QUEUE_URL"
         )
 
+        arq_queue_password: SecretStr | None = Field(
+            None, validation_alias="APP_ARQ_QUEUE_PASSWORD"
+        )
+
         arq_mode: ArqMode = Field(
             ArqMode.production, validation_alias="APP_ARQ_MODE"
         )
 
         @property
         def arq_redis_settings(self) -> RedisSettings:
             """Create a Redis settings instance for arq."""
-            url_parts = urlparse(self.redis_queue_url)
-            redis_settings = RedisSettings(
-                host=url_parts.hostname or "localhost",
-                port=url_parts.port or 6379,
-                database=(
-                    int(url_parts.path.lstrip("/")) if url_parts.path else 0
-                ),
+            return build_arq_redis_settings(
+                self.arq_queue_url, self.arq_queue_password
             )
-            return redis_settings
 
 The `safir.pydantic.EnvRedisDsn` type will automatically incorporate Redis location information from tox-docker.
 See :ref:`pydantic-dsns` for more details.

src/safir/arq.py

@@ -14,6 +14,8 @@
 from arq.connections import ArqRedis, RedisSettings
 from arq.constants import default_queue_name as arq_default_queue_name
 from arq.jobs import Job, JobStatus
+from pydantic import SecretStr
+from pydantic_core import Url
 
 from .datetime import current_datetime
 
@@ -28,6 +30,7 @@
     "ArqQueue",
     "RedisArqQueue",
     "MockArqQueue",
+    "build_arq_redis_settings",
 ]
 
 
@@ -618,3 +621,51 @@ async def set_complete(
             queue_name=queue_name,
         )
         self._job_results[queue_name][job_id] = result_info
+
+
+def build_arq_redis_settings(
+    url: Url, password: SecretStr | None
+) -> RedisSettings:
+    """Construct Redis settings for arq.
+
+    Parameters
+    ----------
+    url
+        Redis DSN.
+    password
+        Password for the Redis connection.
+
+    Returns
+    -------
+    arq.connections.RedisSettings
+        Settings for the arq Redis pool.
+
+    Examples
+    --------
+    This function is normally used from a property in the application
+    configuration. The application should usually use
+    `~safir.pydantic.EnvRedisDsn` as the type for the Redis DSN.
+
+    .. code-block:: python
+
+       from arq.connections import RedisSettings
+       from pydantic_settings import BaseSettings
+       from safir.pydantic import EnvRedisDsn
+
+
+       class Config(BaseSettings):
+           arq_queue_url: EnvRedisDsn
+           arq_queue_password: SecretStr | None
+
+           @property
+           def arq_redis_settings(self) -> RedisSettings:
+               return build_arq_redis_settings(
+                   self.arq_queue_url, self_arq_queue_password
+               )
+    """
+    return RedisSettings(
+        host=url.unicode_host() or "localhost",
+        port=url.port or 6379,
+        database=int(url.path.lstrip("/")) if url.path else 0,
+        password=password.get_secret_value() if password else None,
+    )

tests/arq_test.py

@@ -0,0 +1,27 @@
+"""Tests for arq utility functions.
+
+Most of the arq support code is tested by testing the FastAPI dependency.
+"""
+
+from __future__ import annotations
+
+from pydantic import SecretStr
+from pydantic_core import Url
+
+from safir.arq import build_arq_redis_settings
+
+
+def test_build_arq_redis_settings() -> None:
+    url = Url.build(scheme="redis", host="localhost")
+    settings = build_arq_redis_settings(url, None)
+    assert settings.host == "localhost"
+    assert settings.port == 6379
+    assert settings.database == 0
+    assert settings.password is None
+
+    url = Url.build(scheme="redis", host="example.com", port=7777, path="4")
+    settings = build_arq_redis_settings(url, SecretStr("password"))
+    assert settings.host == "example.com"
+    assert settings.port == 7777
+    assert settings.database == 4
+    assert settings.password == "password"