ss-tools/backend/src/api/routes/assistant/_tool_backup.py

# #region AssistantToolBackup [C:3] [TYPE Module] [SEMANTICS assistant, tool, backup]
# @defgroup AssistantApi Module group.
# @BRIEF Handler for the "run_backup" tool — run backup for environment or specific dashboard.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [TaskManager]

from __future__ import annotations

from typing import Any

from fastapi import HTTPException
from sqlalchemy.orm import Session

from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User

from ._resolvers import (
    _get_environment_name_by_id,
    _resolve_dashboard_id_entity,
    _resolve_env_id,
)
from ._schemas import AssistantAction
from ._tool_registry import _check_any_permission, assistant_tool


# #region handle_run_backup [C:3] [TYPE Function]
# @ingroup AssistantApi
@assistant_tool(
    operation="run_backup",
    domain="backup",
    description="Run backup for environment or specific dashboard by id/slug/title",
    required_entities=["environment"],
    optional_entities=["dashboard_id", "dashboard_ref"],
    risk_level="guarded",
    requires_confirmation=False,
    permission_checks=[
        ("plugin:superset-backup", "EXECUTE"),
        ("plugin:backup", "EXECUTE"),
    ],
)
@belief_scope("run_backup")
async def handle_run_backup(
    intent: dict[str, Any],
    current_user: User,
    task_manager: TaskManager,
    config_manager: ConfigManager,
    db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
    """Run backup for environment or specific dashboard."""
    _check_any_permission(
        current_user,
        [("plugin:superset-backup", "EXECUTE"), ("plugin:backup", "EXECUTE")],
    )
    entities = intent.get("entities", {})
    env_token = entities.get("environment")
    env_id = _resolve_env_id(env_token, config_manager)
    if not env_id:
        raise HTTPException(status_code=400, detail="Missing or unknown environment")
    params: dict[str, Any] = {"environment_id": env_id}
    if entities.get("dashboard_id") or entities.get("dashboard_ref"):
        dashboard_id = await _resolve_dashboard_id_entity(
            entities, config_manager, env_hint=env_token
        )
        if not dashboard_id:
            raise HTTPException(
                status_code=422, detail="Missing dashboard_id/dashboard_ref"
            )
        params["dashboard_ids"] = [dashboard_id]
    task = await task_manager.create_task(
        plugin_id="superset-backup", params=params, user_id=current_user.id
    )
    actions: list[AssistantAction] = [
        AssistantAction(type="open_task", label="Open Task", target=task.id),
        AssistantAction(
            type="open_reports", label="Open Reports", target="/reports"
        ),
    ]
    if entities.get("dashboard_id") or entities.get("dashboard_ref"):
        dashboard_id = await _resolve_dashboard_id_entity(
            entities, config_manager, env_hint=env_token
        )
        if dashboard_id:
            actions.append(
                AssistantAction(
                    type="open_route",
                    label=f"Открыть дашборд в {_get_environment_name_by_id(env_id, config_manager)}",
                    target=f"/dashboards/{dashboard_id}?env_id={env_id}",
                )
            )
            actions.append(
                AssistantAction(
                    type="open_diff", label="Показать Diff", target=str(dashboard_id)
                )
            )
    return (f"Бэкап запущен. task_id={task.id}", task.id, actions)


# #endregion handle_run_backup
# #endregion AssistantToolBackup