WIP: init: build redis url from its parts if available

CHANGES.rst

@@ -9,6 +9,10 @@
 Changes
 =======
 
+Version <next>
+
+- init: build redis url from its parts if available
+
 Version 2.0.0 (released 2024-12-03)
 
 - setup: replace invenio-accounts with flask-login

invenio_cache/ext.py

@@ -35,6 +35,32 @@ def init_app(self, app):
         )
         app.extensions["invenio-cache"] = self
 
+    def build_redis_url(self, app, db=0):
+        """Return the redis connection string if configured or build it from its parts.
+
+        If set, then ``CACHE_REDIS_URL`` will be returned.
+        Otherwise, the URI will be pieced together by the configuration items
+        ``KV_CACHE_{PASSWORD,HOST,PORT,NAME,PROTOCOL}``.
+        If that cannot be done (e.g. because required values are missing), then
+        ``None`` will be returned.
+
+        Note: see: https://docs.celeryq.dev/en/stable/userguide/configuration.html#new-lowercase-settings
+        """
+        if url := app.config.get("CACHE_REDIS_URL", None):
+            return url
+
+        params = {}
+        for config_name in ["HOST", "PORT", "PASSWORD", "PROTOCOL"]:
+            params[config_name] = app.config.get(f"KV_CACHE_{config_name}", None)
+
+        if params.get("HOST") and params.get("PORT"):
+            protocol = params.get("PROTOCOL", "redis")
+            password = f":{params['PASSWORD']}@" if params.get("PASSWORD") else ""
+            cache_url = f"{protocol}://{password}{params['HOST']}:{params['PORT']}/{db}"
+            return cache_url
+
+        return None
+
     def init_config(self, app):
         """Initialize configuration."""
         for k in dir(config):

tests/test_invenio_cache.py

@@ -11,8 +11,8 @@
 from __future__ import absolute_import, print_function
 
 import pkg_resources
+import pytest
 from flask import Flask
-from flask_login import current_user
 from mock import patch
 
 from invenio_cache import (
@@ -81,3 +81,51 @@ def test_callback_no_login(get_distribution):
     """Test callback factory (no flask-login)."""
     get_distribution.side_effect = pkg_resources.DistributionNotFound
     assert _callback_factory(None)() is False
+
+
+@pytest.mark.parametrize(
+    "configs, db, expected_url",
+    [
+        (
+            {
+                "KV_CACHE_HOST": "testhost",
+                "KV_CACHE_PORT": "6379",
+                "KV_CACHE_PASSWORD": "testpassword",
+                "KV_CACHE_PROTOCOL": "redis",
+            },
+            2,
+            "redis://:testpassword@testhost:6379/2",
+        ),
+        (
+            {
+                "KV_CACHE_HOST": "testhost",
+                "KV_CACHE_PORT": "6379",
+                "KV_CACHE_PROTOCOL": "redis",
+            },
+            1,
+            "redis://testhost:6379/1",
+        ),
+        (
+            {"BROKER_URL": "redis://localhost:6379/0"},
+            None,
+            "redis://localhost:6379/0",
+        ),
+        (
+            {"KV_CACHE_URL": "redis://localhost:6379/3"},
+            3,
+            "redis://localhost:6379/3",
+        ),
+        (
+            {},
+            4,
+            "redis://localhost:6379/4",
+        ),
+    ],
+)
+def test_build_redis_url(configs, db, expected_url):
+    """Test building Redis URL."""
+    app = Flask("test_app")
+    assert "CACHE_REDIS_URL" not in app.config
+    app.config.update(configs)
+    InvenioCache(app)
+    assert app.config["CACHE_REDIS_URL"] == expected_url