# Logger initialization
logger = get_logger(__name__)
-# -----------------------------
-# Return a specific config value
-# -----------------------------
+# ---------------------------------------------------------
+# Type mapping for config keys
+# ---------------------------------------------------------
+CONFIG_TYPES = {
+ "external_name": str,
+ "login_max_attempts": int,
+ "login_window_seconds": int,
+}
+
+# ---------------------------------------------------------
+# Runtime cache to avoid repeated DB queries
+# ---------------------------------------------------------
+_config_cache = {}
+
+def invalidate_config(key=None):
+ """Clear cached config entry (or full cache)."""
+ if key:
+ _config_cache.pop(key, None)
+ else:
+ _config_cache.clear()
+
+# ---------------------------------------------------------
+# Return a specific config value (with cache + type casting)
+# ---------------------------------------------------------
def get_config(key):
+ # ---- Cache hit ----
+ if key in _config_cache:
+ return _config_cache[key]
+
+ # ---- Read from DB ----
conn = get_db()
cur = conn.execute("SELECT value FROM config WHERE key = ?", (key,))
row = cur.fetchone()
- return row["value"] if row else None
-# -----------------------------
+ if not row:
+ return None
+
+ raw_value = row["value"]
+
+ # ---- Type casting ----
+ caster = CONFIG_TYPES.get(key, str)
+ try:
+ value = caster(raw_value)
+ except Exception:
+ value = raw_value # fallback safe
+
+ # ---- Save in cache ----
+ _config_cache[key] = value
+ return value
+
+# ---------------------------------------------------------
# Initialize Config DB Table
-# -----------------------------
+# ---------------------------------------------------------
@register_init
def init_db_hosts_table(cur):
- # SETTINGS TABLE
+ # CONFIG TABLE
cur.execute("""
CREATE TABLE config (
key TEXT PRIMARY KEY,
value TEXT
);
""")
- cur.execute("INSERT INTO config (key, value) VALUES (?, ?)", ("domain", settings.DOMAIN))
+
+ # Initial values from settings (as strings in DB)
cur.execute("INSERT INTO config (key, value) VALUES (?, ?)", ("external_name", settings.EXTERNAL_NAME))
+ cur.execute("INSERT INTO config (key, value) VALUES (?, ?)", ("login_max_attempts", str(settings.LOGIN_MAX_ATTEMPTS)))
+ cur.execute("INSERT INTO config (key, value) VALUES (?, ?)", ("login_window_seconds", str(settings.LOGIN_WINDOW_SECONDS)))
logger.info("CONFIG DB: Tables initialized successfully")
import time
# Import local modules
+from backend.db.config import get_config
from backend.security import verify_login, apply_session, close_session
# Import Settings & Logging
now = time.time()
attempts = login_attempts.get(ip, [])
# tieni solo tentativi negli ultimi LOGIN_WINDOW_SECONDS secondi
- attempts = [t for t in attempts if now - t < settings.LOGIN_WINDOW_SECONDS]
+ attempts = [t for t in attempts if now - t < int(get_config("login_window_seconds"))]
- if len(attempts) >= settings.LOGIN_MAX_ATTEMPTS:
+ if len(attempts) >= int(get_config("login_max_attempts")):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail={