insightfy-bms-ms-ums / settings.py
MukeshKapoor25's picture
fix(settings): construct Redis URL from CACHE_URI when CACHE_URL unavailable
3284c84
from __future__ import annotations
from urllib.parse import quote_plus, urlencode
import os
from insightfy_utils.config import load_env
load_env()
SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = os.getenv("ALGORITHM", "HS256")
MONGO_URI = os.getenv('MONGO_URI')
MONGO_DB_NAME = os.getenv("MONGO_DB_NAME", "insightfy-bloom").strip()
def _sanitize_mongo_uri(uri: str | None) -> str | None:
"""Sanitize MongoDB URI to remove invalid options that cause warnings."""
if not uri:
return uri
# Fix double question marks if present
uri = uri.replace("??", "?")
# Remove unsupported URI options that might be present in legacy connection strings
# or from copy-paste errors
if "?" in uri:
base_uri, options_part = uri.split("?", 1)
options = options_part.split("&")
valid_options = []
for opt in options:
# Skip empty options
if not opt:
continue
key = opt.split("=")[0] if "=" in opt else opt
# Filter out known problematic options
# ?ssl (with leading question mark in key) happens if ??ssl was used
# ssl_cert_reqs is a python kwarg, not a valid URI option
if key in ["?ssl", "ssl_cert_reqs"]:
continue
valid_options.append(opt)
return f"{base_uri}?{'&'.join(valid_options)}"
return uri
MONGO_URI = _sanitize_mongo_uri(os.getenv('MONGO_URI'))
CACHE_URL = os.getenv("CACHE_URL") or os.getenv("REDIS_URL")
CACHE_URI = os.getenv('CACHE_URI')
CACHE_K = os.getenv('CACHE_K')
CACHE_DB = int(os.getenv("CACHE_DB", "0"))
if not CACHE_URL and CACHE_URI:
# Build redis:// URL from CACHE_URI (host:port format)
if CACHE_K:
# With password: redis://:password@host:port/db
CACHE_URL = f"redis://:{CACHE_K}@{CACHE_URI}/{CACHE_DB}"
else:
# Without password: redis://host:port/db
CACHE_URL = f"redis://{CACHE_URI}/{CACHE_DB}"