ss-tools/backend/src/api/routes/environments.py

# #region EnvironmentsApi [C:5] [TYPE Module] [SEMANTICS fastapi, environment, api]
# @defgroup Api Module group.
#
# @BRIEF API endpoints for listing environments and their databases.
# @LAYER API
# @RELATION DEPENDS_ON -> [AppDependencies]
# @RELATION DEPENDS_ON -> [SupersetClient]
#
# @INVARIANT  Environment IDs must exist in the configuration.


from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field

from ...core.logger import belief_scope
from ...core.async_superset_client import AsyncSupersetClient
from ...dependencies import get_config_manager, get_scheduler_service, has_permission

router = APIRouter(prefix="/api/environments", tags=["Environments"])


# #region _normalize_superset_env_url [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 ScheduleSchema [TYPE DataClass]
# @ingroup Api
class ScheduleSchema(BaseModel):
    enabled: bool = False
    cron_expression: str = Field(..., pattern=r'^(@(annually|yearly|monthly|weekly|daily|hourly|reboot))|((((\d+,)*\d+|(\d+(\/|-)\d+)|\d+|\*) ?){4,6})$')
# #endregion ScheduleSchema

# #region EnvironmentResponse [TYPE DataClass]
# @ingroup Api
class EnvironmentResponse(BaseModel):
    id: str
    name: str
    url: str
    stage: str = "DEV"
    is_production: bool = False
    backup_schedule: ScheduleSchema | None = None
# #endregion EnvironmentResponse

# #region DatabaseResponse [TYPE DataClass]
# @ingroup Api
class DatabaseResponse(BaseModel):
    uuid: str
    database_name: str
    engine: str | None
# #endregion DatabaseResponse

# #region get_environments [TYPE Function] [SEMANTICS list, environments, config]
# @ingroup Api
# @BRIEF List all configured environments.
# @LAYER API
# @PRE      config_manager is injected via Depends.
# @POST     Returns a list of EnvironmentResponse objects.
@router.get("", response_model=list[EnvironmentResponse])
async def get_environments(
    config_manager=Depends(get_config_manager),
    _ = Depends(has_permission("environments", "READ"))
):
    with belief_scope("get_environments"):
        envs = config_manager.get_environments()
    # Ensure envs is a list
    if not isinstance(envs, list):
        envs = []
    response_items = []
    for e in envs:
        resolved_stage = str(
            getattr(e, "stage", "")
            or ("PROD" if bool(getattr(e, "is_production", False)) else "DEV")
        ).upper()
        response_items.append(
            EnvironmentResponse(
                id=e.id,
                name=e.name,
                url=_normalize_superset_env_url(e.url),
                stage=resolved_stage,
                is_production=(resolved_stage == "PROD"),
                backup_schedule=ScheduleSchema(
                    enabled=e.backup_schedule.enabled,
                    cron_expression=e.backup_schedule.cron_expression
                ) if getattr(e, 'backup_schedule', None) else None
            )
        )
    return response_items
# #endregion get_environments

# #region update_environment_schedule [TYPE Function] [SEMANTICS update, schedule, backup, environment]
# @ingroup Api
# @BRIEF Update backup schedule for an environment.
# @LAYER API
# @PRE      Environment id exists, schedule is valid ScheduleSchema.
# @POST     Backup schedule updated and scheduler reloaded.
@router.put("/{id}/schedule")
async def update_environment_schedule(
    id: str,
    schedule: ScheduleSchema,
    config_manager=Depends(get_config_manager),
    scheduler_service=Depends(get_scheduler_service),
    _ = Depends(has_permission("admin:settings", "WRITE"))
):
    with belief_scope("update_environment_schedule", f"id={id}"):
        envs = config_manager.get_environments()
    env = next((e for e in envs if e.id == id), None)
    if not env:
        raise HTTPException(status_code=404, detail="Environment not found")

    # Update environment config
    env.backup_schedule.enabled = schedule.enabled
    env.backup_schedule.cron_expression = schedule.cron_expression

    config_manager.update_environment(id, env)

    # Refresh scheduler
    scheduler_service.load_schedules()

    return {"message": "Schedule updated successfully"}
# #endregion update_environment_schedule

# #region get_environment_databases [TYPE Function] [SEMANTICS fetch, databases, superset, environment]
# @ingroup Api
# @BRIEF Fetch the list of databases from a specific environment.
# @LAYER API
# @PRE      Environment id exists.
# @POST     Returns a list of database summaries from the environment.
@router.get("/{id}/databases")
async def get_environment_databases(
    id: str,
    config_manager=Depends(get_config_manager),
    _ = Depends(has_permission("admin:settings", "READ"))
):
    with belief_scope("get_environment_databases", f"id={id}"):
        envs = config_manager.get_environments()
    env = next((e for e in envs if e.id == id), None)
    if not env:
        raise HTTPException(status_code=404, detail="Environment not found")

    try:
        # Initialize AsyncSupersetClient from environment config
        client = AsyncSupersetClient(env)
        return await client.get_databases_summary()
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to fetch databases: {e!s}")
# #endregion get_environment_databases

# #endregion EnvironmentsApi