busya
b6f46fe9f5
Replace [Entry]/[Exit]/[Action]/[COHERENCE:OK] with [REASON]/[REFLECT]/[EXPLORE] in belief_scope context manager, BeliefFormatter, and 34 source files. Per molecular-cot-logging skill invariants.
602 lines
21 KiB
Python
Executable File
602 lines
21 KiB
Python
Executable File
# #region SettingsRouter [C:3] [TYPE Module] [SEMANTICS settings, api, router, fastapi]
|
|
#
|
|
# @BRIEF Provides API endpoints for managing application settings and Superset environments.
|
|
# @LAYER: API
|
|
# @RELATION DEPENDS_ON -> [ConfigManager]
|
|
# @RELATION DEPENDS_ON -> [get_config_manager]
|
|
# @RELATION DEPENDS_ON -> [has_permission]
|
|
#
|
|
# @INVARIANT: All settings changes must be persisted via ConfigManager.
|
|
# @PUBLIC_API: router
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from typing import List
|
|
from pydantic import BaseModel
|
|
from ...core.config_models import AppConfig, Environment, GlobalSettings, LoggingConfig
|
|
from ...models.storage import StorageConfig
|
|
from ...dependencies import get_config_manager, has_permission
|
|
from ...core.config_manager import ConfigManager
|
|
from ...core.logger import logger, belief_scope
|
|
from ...core.superset_client import SupersetClient
|
|
from ...services.llm_prompt_templates import normalize_llm_settings
|
|
from ...models.llm import ValidationPolicy
|
|
from ...models.config import AppConfigRecord
|
|
from ...schemas.settings import (
|
|
ValidationPolicyCreate,
|
|
ValidationPolicyUpdate,
|
|
ValidationPolicyResponse,
|
|
)
|
|
from ...core.database import get_db
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
# #region LoggingConfigResponse [C:1] [TYPE Class] [SEMANTICS logging, config, response]
|
|
# @BRIEF Response model for logging configuration with current task log level.
|
|
class LoggingConfigResponse(BaseModel):
|
|
level: str
|
|
task_log_level: str
|
|
enable_belief_state: bool
|
|
|
|
|
|
# #endregion LoggingConfigResponse
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# #region _normalize_superset_env_url [C:1] [TYPE Function]
|
|
# @BRIEF Canonicalize Superset environment URL to base host/path without trailing /api/v1.
|
|
# @PRE: raw_url can be empty.
|
|
# @POST: Returns normalized base URL.
|
|
def _normalize_superset_env_url(raw_url: str) -> str:
|
|
normalized = str(raw_url or "").strip().rstrip("/")
|
|
if normalized.lower().endswith("/api/v1"):
|
|
normalized = normalized[: -len("/api/v1")]
|
|
return normalized.rstrip("/")
|
|
|
|
|
|
# #endregion _normalize_superset_env_url
|
|
|
|
|
|
# #region _validate_superset_connection_fast [C:2] [TYPE Function]
|
|
# @BRIEF Run lightweight Superset connectivity validation without full pagination scan.
|
|
# @PRE: env contains valid URL and credentials.
|
|
# @POST: Raises on auth/API failures; returns None on success.
|
|
def _validate_superset_connection_fast(env: Environment) -> None:
|
|
client = SupersetClient(env)
|
|
# 1) Explicit auth check
|
|
client.authenticate()
|
|
# 2) Single lightweight API call to ensure read access
|
|
client.get_dashboards_page(
|
|
query={
|
|
"page": 0,
|
|
"page_size": 1,
|
|
"columns": ["id"],
|
|
}
|
|
)
|
|
|
|
|
|
# #endregion _validate_superset_connection_fast
|
|
|
|
|
|
# #region get_settings [C:2] [TYPE Function]
|
|
# @BRIEF Retrieves all application settings.
|
|
# @PRE: Config manager is available.
|
|
# @POST: Returns masked AppConfig.
|
|
@router.get("", response_model=AppConfig)
|
|
async def get_settings(
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
_=Depends(has_permission("admin:settings", "READ")),
|
|
):
|
|
with belief_scope("get_settings"):
|
|
logger.info("[get_settings][REASON] Fetching all settings")
|
|
config = config_manager.get_config().copy(deep=True)
|
|
config.settings.llm = normalize_llm_settings(config.settings.llm)
|
|
# Mask passwords
|
|
for env in config.environments:
|
|
if env.password:
|
|
env.password = "********"
|
|
return config
|
|
|
|
|
|
# #endregion get_settings
|
|
|
|
|
|
# #region get_features [C:1] [TYPE Function]
|
|
# @BRIEF Public endpoint returning feature flags for frontend sidebar filtering.
|
|
# @RATIONALE: Unauthenticated because sidebar filtering must work for all users, not just admins.
|
|
# @PRE: Config manager is available.
|
|
# @POST: Returns dict with dataset_review and health_monitor booleans.
|
|
@router.get("/features")
|
|
async def get_features(
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
):
|
|
return config_manager.get_config().settings.features.model_dump()
|
|
|
|
|
|
# #endregion get_features
|
|
|
|
|
|
# #region update_global_settings [C:2] [TYPE Function]
|
|
# @BRIEF Updates global application settings.
|
|
# @PRE: New settings are provided.
|
|
# @POST: Global settings are updated.
|
|
@router.patch("/global", response_model=GlobalSettings)
|
|
async def update_global_settings(
|
|
settings: GlobalSettings,
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
_=Depends(has_permission("admin:settings", "WRITE")),
|
|
):
|
|
with belief_scope("update_global_settings"):
|
|
logger.info("[update_global_settings][REASON] Updating global settings")
|
|
|
|
config_manager.update_global_settings(settings)
|
|
return settings
|
|
|
|
|
|
# #endregion update_global_settings
|
|
|
|
|
|
# #region get_storage_settings [C:2] [TYPE Function]
|
|
# @BRIEF Retrieves storage-specific settings.
|
|
@router.get("/storage", response_model=StorageConfig)
|
|
async def get_storage_settings(
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
_=Depends(has_permission("admin:settings", "READ")),
|
|
):
|
|
with belief_scope("get_storage_settings"):
|
|
return config_manager.get_config().settings.storage
|
|
|
|
|
|
# #endregion get_storage_settings
|
|
|
|
|
|
# #region update_storage_settings [C:2] [TYPE Function]
|
|
# @BRIEF Updates storage-specific settings.
|
|
# @POST: Storage settings are updated and saved.
|
|
@router.put("/storage", response_model=StorageConfig)
|
|
async def update_storage_settings(
|
|
storage: StorageConfig,
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
_=Depends(has_permission("admin:settings", "WRITE")),
|
|
):
|
|
with belief_scope("update_storage_settings"):
|
|
is_valid, message = config_manager.validate_path(storage.root_path)
|
|
if not is_valid:
|
|
raise HTTPException(status_code=400, detail=message)
|
|
|
|
settings = config_manager.get_config().settings
|
|
settings.storage = storage
|
|
config_manager.update_global_settings(settings)
|
|
return config_manager.get_config().settings.storage
|
|
|
|
|
|
# #endregion update_storage_settings
|
|
|
|
|
|
# #region get_environments [C:2] [TYPE Function]
|
|
# @BRIEF Lists all configured Superset environments.
|
|
# @PRE: Config manager is available.
|
|
# @POST: Returns list of environments.
|
|
@router.get("/environments", response_model=List[Environment])
|
|
async def get_environments(
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
_=Depends(has_permission("admin:settings", "READ")),
|
|
):
|
|
with belief_scope("get_environments"):
|
|
logger.info("[get_environments][REASON] Fetching environments")
|
|
environments = config_manager.get_environments()
|
|
return [
|
|
env.copy(update={"url": _normalize_superset_env_url(env.url)})
|
|
for env in environments
|
|
]
|
|
|
|
|
|
# #endregion get_environments
|
|
|
|
|
|
# #region add_environment [C:2] [TYPE Function]
|
|
# @BRIEF Adds a new Superset environment.
|
|
# @PRE: Environment data is valid and reachable.
|
|
# @POST: Environment is added to config.
|
|
@router.post("/environments", response_model=Environment)
|
|
async def add_environment(
|
|
env: Environment,
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
_=Depends(has_permission("admin:settings", "WRITE")),
|
|
):
|
|
with belief_scope("add_environment"):
|
|
logger.info(f"[add_environment][REASON] Adding environment {env.id}")
|
|
env = env.copy(update={"url": _normalize_superset_env_url(env.url)})
|
|
|
|
# Validate connection before adding (fast path)
|
|
try:
|
|
_validate_superset_connection_fast(env)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"[add_environment][Coherence:Failed] Connection validation failed: {e}"
|
|
)
|
|
raise HTTPException(
|
|
status_code=400, detail=f"Connection validation failed: {e}"
|
|
)
|
|
|
|
config_manager.add_environment(env)
|
|
return env
|
|
|
|
|
|
# #endregion add_environment
|
|
|
|
|
|
# #region update_environment [C:2] [TYPE Function]
|
|
# @BRIEF Updates an existing Superset environment.
|
|
# @PRE: ID and valid environment data are provided.
|
|
# @POST: Environment is updated in config.
|
|
@router.put("/environments/{id}", response_model=Environment)
|
|
async def update_environment(
|
|
id: str,
|
|
env: Environment,
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
):
|
|
with belief_scope("update_environment"):
|
|
logger.info(f"[update_environment][REASON] Updating environment {id}")
|
|
|
|
env = env.copy(update={"url": _normalize_superset_env_url(env.url)})
|
|
|
|
# If password is masked, we need the real one for validation
|
|
env_to_validate = env.copy(deep=True)
|
|
if env_to_validate.password == "********":
|
|
old_env = next(
|
|
(e for e in config_manager.get_environments() if e.id == id), None
|
|
)
|
|
if old_env:
|
|
env_to_validate.password = old_env.password
|
|
|
|
# Validate connection before updating (fast path)
|
|
try:
|
|
_validate_superset_connection_fast(env_to_validate)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"[update_environment][Coherence:Failed] Connection validation failed: {e}"
|
|
)
|
|
raise HTTPException(
|
|
status_code=400, detail=f"Connection validation failed: {e}"
|
|
)
|
|
|
|
if config_manager.update_environment(id, env):
|
|
return env
|
|
raise HTTPException(status_code=404, detail=f"Environment {id} not found")
|
|
|
|
|
|
# #endregion update_environment
|
|
|
|
|
|
# #region delete_environment [C:2] [TYPE Function]
|
|
# @BRIEF Deletes a Superset environment.
|
|
# @PRE: ID is provided.
|
|
# @POST: Environment is removed from config.
|
|
@router.delete("/environments/{id}")
|
|
async def delete_environment(
|
|
id: str, config_manager: ConfigManager = Depends(get_config_manager)
|
|
):
|
|
with belief_scope("delete_environment"):
|
|
logger.info(f"[delete_environment][REASON] Deleting environment {id}")
|
|
config_manager.delete_environment(id)
|
|
return {"message": f"Environment {id} deleted"}
|
|
|
|
|
|
# #endregion delete_environment
|
|
|
|
|
|
# #region test_environment_connection [C:2] [TYPE Function]
|
|
# @BRIEF Tests the connection to a Superset environment.
|
|
# @PRE: ID is provided.
|
|
# @POST: Returns success or error status.
|
|
@router.post("/environments/{id}/test")
|
|
async def test_environment_connection(
|
|
id: str, config_manager: ConfigManager = Depends(get_config_manager)
|
|
):
|
|
with belief_scope("test_environment_connection"):
|
|
logger.info(f"[test_environment_connection][REASON] Testing environment {id}")
|
|
|
|
# Find environment
|
|
env = next((e for e in config_manager.get_environments() if e.id == id), None)
|
|
if not env:
|
|
raise HTTPException(status_code=404, detail=f"Environment {id} not found")
|
|
|
|
try:
|
|
_validate_superset_connection_fast(env)
|
|
|
|
logger.info(
|
|
f"[test_environment_connection][Coherence:OK] Connection successful for {id}"
|
|
)
|
|
return {"status": "success", "message": "Connection successful"}
|
|
except Exception as e:
|
|
logger.error(
|
|
f"[test_environment_connection][Coherence:Failed] Connection failed for {id}: {e}"
|
|
)
|
|
return {"status": "error", "message": str(e)}
|
|
|
|
|
|
# #endregion test_environment_connection
|
|
|
|
|
|
# #region get_logging_config [C:2] [TYPE Function]
|
|
# @BRIEF Retrieves current logging configuration.
|
|
# @PRE: Config manager is available.
|
|
# @POST: Returns logging configuration.
|
|
@router.get("/logging", response_model=LoggingConfigResponse)
|
|
async def get_logging_config(
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
_=Depends(has_permission("admin:settings", "READ")),
|
|
):
|
|
with belief_scope("get_logging_config"):
|
|
logging_config = config_manager.get_config().settings.logging
|
|
return LoggingConfigResponse(
|
|
level=logging_config.level,
|
|
task_log_level=logging_config.task_log_level,
|
|
enable_belief_state=logging_config.enable_belief_state,
|
|
)
|
|
|
|
|
|
# #endregion get_logging_config
|
|
|
|
|
|
# #region update_logging_config [C:2] [TYPE Function]
|
|
# @BRIEF Updates logging configuration.
|
|
# @PRE: New logging config is provided.
|
|
# @POST: Logging configuration is updated and saved.
|
|
@router.patch("/logging", response_model=LoggingConfigResponse)
|
|
async def update_logging_config(
|
|
config: LoggingConfig,
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
_=Depends(has_permission("admin:settings", "WRITE")),
|
|
):
|
|
with belief_scope("update_logging_config"):
|
|
logger.info(
|
|
f"[update_logging_config][REASON] Updating logging config: level={config.level}, task_log_level={config.task_log_level}"
|
|
)
|
|
|
|
# Get current settings and update logging config
|
|
settings = config_manager.get_config().settings
|
|
settings.logging = config
|
|
config_manager.update_global_settings(settings)
|
|
|
|
return LoggingConfigResponse(
|
|
level=config.level,
|
|
task_log_level=config.task_log_level,
|
|
enable_belief_state=config.enable_belief_state,
|
|
)
|
|
|
|
|
|
# #endregion update_logging_config
|
|
|
|
|
|
# #region ConsolidatedSettingsResponse [C:1] [TYPE Class]
|
|
# @BRIEF Response model for consolidated application settings.
|
|
class ConsolidatedSettingsResponse(BaseModel):
|
|
environments: List[dict]
|
|
connections: List[dict]
|
|
llm: dict
|
|
llm_providers: List[dict]
|
|
logging: dict
|
|
storage: dict
|
|
notifications: dict = {}
|
|
features: dict = {}
|
|
|
|
|
|
# #endregion ConsolidatedSettingsResponse
|
|
|
|
|
|
# #region get_consolidated_settings [C:4] [TYPE Function]
|
|
# @BRIEF Retrieves all settings categories in a single call.
|
|
# @PRE: Config manager is available and the caller holds admin settings read permission.
|
|
# @POST: Returns consolidated settings, provider metadata, and persisted notification payload in one stable response.
|
|
# @SIDE_EFFECT: Opens one database session to read LLM providers and config-backed notification payload, then closes it.
|
|
# @DATA_CONTRACT: Input[ConfigManager] -> Output[ConsolidatedSettingsResponse]
|
|
# @RELATION DEPENDS_ON -> [ConfigManager]
|
|
# @RELATION DEPENDS_ON -> [LLMProviderService]
|
|
# @RELATION DEPENDS_ON -> [AppConfigRecord]
|
|
# @RELATION DEPENDS_ON -> [SessionLocal]
|
|
# @RELATION DEPENDS_ON -> [has_permission]
|
|
# @RELATION DEPENDS_ON -> [normalize_llm_settings]
|
|
@router.get("/consolidated", response_model=ConsolidatedSettingsResponse)
|
|
async def get_consolidated_settings(
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
_=Depends(has_permission("admin:settings", "READ")),
|
|
):
|
|
with belief_scope("get_consolidated_settings"):
|
|
logger.reason("Fetching consolidated settings payload")
|
|
|
|
config = config_manager.get_config()
|
|
|
|
from ...services.llm_provider import LLMProviderService
|
|
from ...core.database import SessionLocal
|
|
|
|
db = SessionLocal()
|
|
notifications_payload = {}
|
|
try:
|
|
llm_service = LLMProviderService(db)
|
|
providers = llm_service.get_all_providers()
|
|
llm_providers_list = [
|
|
{
|
|
"id": p.id,
|
|
"provider_type": p.provider_type,
|
|
"name": p.name,
|
|
"base_url": p.base_url,
|
|
"api_key": "********",
|
|
"default_model": p.default_model,
|
|
"is_active": p.is_active,
|
|
}
|
|
for p in providers
|
|
]
|
|
|
|
config_record = (
|
|
db.query(AppConfigRecord).filter(AppConfigRecord.id == "global").first()
|
|
)
|
|
if config_record and isinstance(config_record.payload, dict):
|
|
notifications_payload = (
|
|
config_record.payload.get("notifications", {}) or {}
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
normalized_llm = normalize_llm_settings(config.settings.llm)
|
|
|
|
response_payload = ConsolidatedSettingsResponse(
|
|
environments=[env.dict() for env in config.environments],
|
|
connections=config.settings.connections,
|
|
llm=normalized_llm,
|
|
llm_providers=llm_providers_list,
|
|
logging=config.settings.logging.dict(),
|
|
storage=config.settings.storage.dict(),
|
|
notifications=notifications_payload,
|
|
features=config.settings.features.model_dump(),
|
|
)
|
|
logger.reflect(
|
|
"Consolidated settings payload assembled",
|
|
extra={
|
|
"environment_count": len(response_payload.environments),
|
|
"provider_count": len(response_payload.llm_providers),
|
|
},
|
|
)
|
|
return response_payload
|
|
|
|
|
|
# #endregion get_consolidated_settings
|
|
|
|
|
|
# #region update_consolidated_settings [C:2] [TYPE Function]
|
|
# @BRIEF Bulk update application settings from the consolidated view.
|
|
# @PRE: User has admin permissions, config is valid.
|
|
# @POST: Settings are updated and saved via ConfigManager.
|
|
@router.patch("/consolidated")
|
|
async def update_consolidated_settings(
|
|
settings_patch: dict,
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
_=Depends(has_permission("admin:settings", "WRITE")),
|
|
):
|
|
with belief_scope("update_consolidated_settings"):
|
|
logger.info(
|
|
"[update_consolidated_settings][REASON] Applying consolidated settings patch"
|
|
)
|
|
|
|
current_config = config_manager.get_config()
|
|
current_settings = current_config.settings
|
|
|
|
# Update connections if provided
|
|
if "connections" in settings_patch:
|
|
current_settings.connections = settings_patch["connections"]
|
|
|
|
# Update LLM if provided
|
|
if "llm" in settings_patch:
|
|
current_settings.llm = normalize_llm_settings(settings_patch["llm"])
|
|
|
|
# Update Logging if provided
|
|
if "logging" in settings_patch:
|
|
current_settings.logging = LoggingConfig(**settings_patch["logging"])
|
|
|
|
# Update Storage if provided
|
|
if "storage" in settings_patch:
|
|
new_storage = StorageConfig(**settings_patch["storage"])
|
|
is_valid, message = config_manager.validate_path(new_storage.root_path)
|
|
if not is_valid:
|
|
raise HTTPException(status_code=400, detail=message)
|
|
current_settings.storage = new_storage
|
|
|
|
# Update Features if provided
|
|
if "features" in settings_patch:
|
|
from ...core.config_models import FeaturesConfig
|
|
|
|
current_settings.features = FeaturesConfig(**settings_patch["features"])
|
|
|
|
if "notifications" in settings_patch:
|
|
payload = config_manager.get_payload()
|
|
payload["notifications"] = settings_patch["notifications"]
|
|
config_manager.save_config(payload)
|
|
|
|
config_manager.update_global_settings(current_settings)
|
|
return {"status": "success", "message": "Settings updated"}
|
|
|
|
|
|
# #endregion update_consolidated_settings
|
|
|
|
|
|
# #region get_validation_policies [C:2] [TYPE Function]
|
|
# @BRIEF Lists all validation policies.
|
|
@router.get("/automation/policies", response_model=List[ValidationPolicyResponse])
|
|
async def get_validation_policies(
|
|
db: Session = Depends(get_db), _=Depends(has_permission("admin:settings", "READ"))
|
|
):
|
|
with belief_scope("get_validation_policies"):
|
|
return db.query(ValidationPolicy).all()
|
|
|
|
|
|
# #endregion get_validation_policies
|
|
|
|
|
|
# #region create_validation_policy [C:2] [TYPE Function]
|
|
# @BRIEF Creates a new validation policy.
|
|
@router.post("/automation/policies", response_model=ValidationPolicyResponse)
|
|
async def create_validation_policy(
|
|
policy: ValidationPolicyCreate,
|
|
db: Session = Depends(get_db),
|
|
_=Depends(has_permission("admin:settings", "WRITE")),
|
|
):
|
|
with belief_scope("create_validation_policy"):
|
|
db_policy = ValidationPolicy(**policy.dict())
|
|
db.add(db_policy)
|
|
db.commit()
|
|
db.refresh(db_policy)
|
|
return db_policy
|
|
|
|
|
|
# #endregion create_validation_policy
|
|
|
|
|
|
# #region update_validation_policy [C:2] [TYPE Function]
|
|
# @BRIEF Updates an existing validation policy.
|
|
@router.patch("/automation/policies/{id}", response_model=ValidationPolicyResponse)
|
|
async def update_validation_policy(
|
|
id: str,
|
|
policy: ValidationPolicyUpdate,
|
|
db: Session = Depends(get_db),
|
|
_=Depends(has_permission("admin:settings", "WRITE")),
|
|
):
|
|
with belief_scope("update_validation_policy"):
|
|
db_policy = db.query(ValidationPolicy).filter(ValidationPolicy.id == id).first()
|
|
if not db_policy:
|
|
raise HTTPException(status_code=404, detail="Policy not found")
|
|
|
|
update_data = policy.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_policy, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_policy)
|
|
return db_policy
|
|
|
|
|
|
# #endregion update_validation_policy
|
|
|
|
|
|
# #region delete_validation_policy [C:2] [TYPE Function]
|
|
# @BRIEF Deletes a validation policy.
|
|
@router.delete("/automation/policies/{id}")
|
|
async def delete_validation_policy(
|
|
id: str,
|
|
db: Session = Depends(get_db),
|
|
_=Depends(has_permission("admin:settings", "WRITE")),
|
|
):
|
|
with belief_scope("delete_validation_policy"):
|
|
db_policy = db.query(ValidationPolicy).filter(ValidationPolicy.id == id).first()
|
|
if not db_policy:
|
|
raise HTTPException(status_code=404, detail="Policy not found")
|
|
|
|
db.delete(db_policy)
|
|
db.commit()
|
|
return {"message": "Policy deleted"}
|
|
|
|
|
|
# #endregion delete_validation_policy
|
|
|
|
# #endregion SettingsRouter
|