app/core/redis_url.py

"""Build Redis URLs with optional AUTH password (R-055)."""

from __future__ import annotations

from urllib.parse import urlparse, urlunparse


def redis_url_with_auth(url: str, password: str) -> str:
    """Inject password into redis:// URL when not already present."""
    if not password:
        return url
    parsed = urlparse(url)
    if parsed.scheme not in ("redis", "rediss"):
        return url
    if parsed.password or (parsed.username and "@" in url):
        return url
    host = parsed.hostname or "localhost"
    port = f":{parsed.port}" if parsed.port else ""
    netloc = f":{password}@{host}{port}"
    return urlunparse(parsed._replace(netloc=netloc))


def redis_url_has_auth(url: str) -> bool:
    """True when URL includes a non-empty Redis password."""
    parsed = urlparse(url)
    return bool(parsed.password)