daily-briefing/server/services/settings_service.py

298 lines
9.8 KiB
Python
Raw Normal View History

"""Database-backed settings, integrations, and user management."""
from __future__ import annotations
import json
import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from server.db import get_pool
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Admin User
# ---------------------------------------------------------------------------
async def get_admin_user() -> Optional[Dict[str, Any]]:
"""Return the admin user row, or None if not yet created."""
pool = await get_pool()
async with pool.acquire() as conn:
row = await conn.fetchrow("SELECT * FROM admin_user LIMIT 1")
return dict(row) if row else None
async def create_admin_user(username: str, password_hash: str) -> None:
"""Insert the initial admin user."""
pool = await get_pool()
async with pool.acquire() as conn:
await conn.execute(
"INSERT INTO admin_user (username, password_hash) VALUES ($1, $2)",
username,
password_hash,
)
logger.info("Admin user '%s' created", username)
async def update_admin_password(user_id: int, password_hash: str) -> None:
"""Update the admin user's password."""
pool = await get_pool()
async with pool.acquire() as conn:
await conn.execute(
"UPDATE admin_user SET password_hash = $1, updated_at = NOW() WHERE id = $2",
password_hash,
user_id,
)
logger.info("Admin password updated (user_id=%d)", user_id)
# ---------------------------------------------------------------------------
# App Settings (key/value)
# ---------------------------------------------------------------------------
async def get_all_settings() -> Dict[str, Any]:
"""Return all settings as a dict, casting values to their declared type."""
pool = await get_pool()
async with pool.acquire() as conn:
rows = await conn.fetch("SELECT * FROM app_settings ORDER BY category, key")
result: Dict[str, Any] = {}
for row in rows:
result[row["key"]] = {
"value": _cast_value(row["value"], row["value_type"]),
"value_type": row["value_type"],
"category": row["category"],
"label": row["label"],
"description": row["description"],
}
return result
async def get_setting(key: str) -> Optional[Any]:
"""Return a single setting's typed value, or None."""
pool = await get_pool()
async with pool.acquire() as conn:
row = await conn.fetchrow("SELECT value, value_type FROM app_settings WHERE key = $1", key)
if row is None:
return None
return _cast_value(row["value"], row["value_type"])
async def set_setting(
key: str,
value: Any,
value_type: str = "string",
category: str = "general",
label: str = "",
description: str = "",
) -> None:
"""Upsert a single setting."""
pool = await get_pool()
str_value = json.dumps(value) if value_type == "json" else str(value)
async with pool.acquire() as conn:
await conn.execute(
"""
INSERT INTO app_settings (key, value, value_type, category, label, description, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, NOW())
ON CONFLICT (key) DO UPDATE SET
value = EXCLUDED.value,
value_type = EXCLUDED.value_type,
category = EXCLUDED.category,
label = EXCLUDED.label,
description = EXCLUDED.description,
updated_at = NOW()
""",
key, str_value, value_type, category, label, description,
)
async def bulk_set_settings(settings_dict: Dict[str, Any]) -> None:
"""Bulk upsert settings from a flat key→value dict."""
pool = await get_pool()
async with pool.acquire() as conn:
async with conn.transaction():
for key, val in settings_dict.items():
str_val = str(val)
await conn.execute(
"""
UPDATE app_settings SET value = $1, updated_at = NOW()
WHERE key = $2
""",
str_val, key,
)
def _cast_value(raw: str, value_type: str) -> Any:
"""Cast a stored string value to its declared type."""
if value_type == "int":
try:
return int(raw)
except (ValueError, TypeError):
return 0
elif value_type == "bool":
return raw.lower() in ("1", "true", "yes")
elif value_type == "json":
try:
return json.loads(raw)
except (json.JSONDecodeError, TypeError):
return raw
return raw
# ---------------------------------------------------------------------------
# Integrations
# ---------------------------------------------------------------------------
async def get_integrations() -> List[Dict[str, Any]]:
"""Return all integration configs."""
pool = await get_pool()
async with pool.acquire() as conn:
rows = await conn.fetch(
"SELECT * FROM integrations ORDER BY display_order, type"
)
return [_integration_to_dict(row) for row in rows]
async def get_integration(type_name: str) -> Optional[Dict[str, Any]]:
"""Return a single integration by type name."""
pool = await get_pool()
async with pool.acquire() as conn:
row = await conn.fetchrow(
"SELECT * FROM integrations WHERE type = $1", type_name
)
return _integration_to_dict(row) if row else None
async def upsert_integration(
type_name: str,
name: str,
config: Dict[str, Any],
enabled: bool = True,
display_order: int = 0,
) -> Dict[str, Any]:
"""Insert or update an integration config."""
pool = await get_pool()
config_json = json.dumps(config)
async with pool.acquire() as conn:
row = await conn.fetchrow(
"""
INSERT INTO integrations (type, name, config, enabled, display_order, updated_at)
VALUES ($1, $2, $3::jsonb, $4, $5, NOW())
ON CONFLICT (type) DO UPDATE SET
name = EXCLUDED.name,
config = EXCLUDED.config,
enabled = EXCLUDED.enabled,
display_order = EXCLUDED.display_order,
updated_at = NOW()
RETURNING *
""",
type_name, name, config_json, enabled, display_order,
)
return _integration_to_dict(row)
async def toggle_integration(type_name: str, enabled: bool) -> None:
"""Enable or disable an integration."""
pool = await get_pool()
async with pool.acquire() as conn:
await conn.execute(
"UPDATE integrations SET enabled = $1, updated_at = NOW() WHERE type = $2",
enabled, type_name,
)
def _integration_to_dict(row: Any) -> Dict[str, Any]:
"""Convert an integration row to a dict."""
d = dict(row)
# Ensure config is a dict (asyncpg returns JSONB as dict already)
if isinstance(d.get("config"), str):
d["config"] = json.loads(d["config"])
# Convert datetimes
for k in ("created_at", "updated_at"):
if k in d and d[k] is not None:
d[k] = d[k].isoformat()
return d
# ---------------------------------------------------------------------------
# MQTT Subscriptions
# ---------------------------------------------------------------------------
async def get_mqtt_subscriptions() -> List[Dict[str, Any]]:
"""Return all MQTT subscriptions."""
pool = await get_pool()
async with pool.acquire() as conn:
rows = await conn.fetch(
"SELECT * FROM mqtt_subscriptions ORDER BY display_order, id"
)
return [_sub_to_dict(row) for row in rows]
async def create_mqtt_subscription(
topic_pattern: str,
display_name: str = "",
category: str = "other",
unit: str = "",
widget_type: str = "value",
enabled: bool = True,
display_order: int = 0,
) -> Dict[str, Any]:
"""Create a new MQTT subscription."""
pool = await get_pool()
async with pool.acquire() as conn:
row = await conn.fetchrow(
"""
INSERT INTO mqtt_subscriptions
(topic_pattern, display_name, category, unit, widget_type, enabled, display_order)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING *
""",
topic_pattern, display_name, category, unit, widget_type, enabled, display_order,
)
return _sub_to_dict(row)
async def update_mqtt_subscription(sub_id: int, **fields: Any) -> Optional[Dict[str, Any]]:
"""Update specific fields of an MQTT subscription."""
pool = await get_pool()
allowed = {"topic_pattern", "display_name", "category", "unit", "widget_type", "enabled", "display_order"}
updates = {k: v for k, v in fields.items() if k in allowed}
if not updates:
return None
set_parts = []
params = []
for i, (k, v) in enumerate(updates.items(), start=1):
set_parts.append(f"{k} = ${i}")
params.append(v)
params.append(sub_id)
set_clause = ", ".join(set_parts)
async with pool.acquire() as conn:
row = await conn.fetchrow(
f"UPDATE mqtt_subscriptions SET {set_clause}, updated_at = NOW() "
f"WHERE id = ${len(params)} RETURNING *",
*params,
)
return _sub_to_dict(row) if row else None
async def delete_mqtt_subscription(sub_id: int) -> bool:
"""Delete an MQTT subscription. Returns True if deleted."""
pool = await get_pool()
async with pool.acquire() as conn:
result = await conn.execute(
"DELETE FROM mqtt_subscriptions WHERE id = $1", sub_id
)
return result == "DELETE 1"
def _sub_to_dict(row: Any) -> Dict[str, Any]:
d = dict(row)
for k in ("created_at", "updated_at"):
if k in d and d[k] is not None:
d[k] = d[k].isoformat()
return d