# #region GitPluginModule [C:4] [TYPE Module] [SEMANTICS git, versioning, deploy, sync, superset, backup, transactional] # # @BRIEF Предоставляет плагин для версионирования и развертывания дашбордов Superset. # @LAYER Plugin # @RELATION INHERITS -> [PluginBase] # @RELATION DEPENDS_ON -> [GitService] # @RELATION DEPENDS_ON -> [SupersetClient] # @RELATION DEPENDS_ON -> [ConfigManager] # @RELATION DEPENDS_ON -> [TaskContext] # # @INVARIANT Все операции с Git должны выполняться через GitService. # @CONSTRAINT: Плагин работает только с распакованными YAML-экспортами Superset. # @INVARIANT _handle_sync сохраняет backup управляемых директорий перед удалением; # при ошибке распаковки backup восстанавливается. import io import os from pathlib import Path import shutil import time from typing import Any import zipfile from src.core.config_manager import ConfigManager from src.core.logger import belief_scope, logger as app_logger from src.core.plugin_base import PluginBase from src.core.superset_client import SupersetClient from src.core.task_manager.context import TaskContext from src.services.git_service import GitService # #region GitPlugin [TYPE Class] # @BRIEF Реализация плагина Git Integration для управления версиями дашбордов. class GitPlugin(PluginBase): # region __init__ [TYPE Function] # @PURPOSE: Инициализирует плагин и его зависимости. # @PRE shared config_manager доступен через src.dependencies. # @POST Инициализированы git_service и config_manager. def __init__(self): with belief_scope("GitPlugin.__init__"): app_logger.info("Initializing GitPlugin.") self.git_service = GitService() # Используем shared config_manager из dependencies try: from src.dependencies import config_manager self.config_manager = config_manager app_logger.info("GitPlugin initialized using shared config_manager.") except Exception as exc: app_logger.error(f"GitPlugin: failed to get shared config_manager: {exc}") self.config_manager = ConfigManager() app_logger.info("GitPlugin initialized with fallback ConfigManager.") # endregion __init__ @property # region id [C:1] [TYPE Function] def id(self) -> str: return "git-integration" # endregion id @property # region name [C:1] [TYPE Function] def name(self) -> str: return "Git Integration" # endregion name @property # region description [C:1] [TYPE Function] def description(self) -> str: return "Version control for Superset dashboards" # endregion description @property # region version [C:1] [TYPE Function] def version(self) -> str: return "0.1.0" # endregion version @property # region ui_route [C:1] [TYPE Function] def ui_route(self) -> str: return "/git" # endregion ui_route # region get_schema [TYPE Function] # @PURPOSE: Возвращает JSON-схему параметров для выполнения задач плагина. # @PRE GitPlugin is initialized. # @POST Returns a JSON schema dictionary. # @RETURN Dict[str, Any] - Схема параметров. def get_schema(self) -> dict[str, Any]: with belief_scope("GitPlugin.get_schema"): return { "type": "object", "properties": { "operation": {"type": "string", "enum": ["sync", "deploy", "history"]}, "dashboard_id": {"type": "integer"}, "environment_id": {"type": "string"}, "source_env_id": {"type": "string"} }, "required": ["operation", "dashboard_id"] } # endregion get_schema # region initialize [C:1] [TYPE Function] async def initialize(self): app_logger.reason("Initializing Git Integration Plugin logic.", extra={"src": "GitPlugin.initialize"}) # endregion initialize # region execute [C:3] [TYPE Function] # @PURPOSE: Main task executor with TaskContext support. # @RELATION CALLS -> self._handle_sync # @RELATION CALLS -> self._handle_deploy async def execute(self, task_data: dict[str, Any], context: TaskContext | None = None) -> dict[str, Any]: with belief_scope("GitPlugin.execute"): operation = task_data.get("operation") dashboard_id = task_data.get("dashboard_id") # Use TaskContext logger if available, otherwise fall back to app_logger log = context.logger if context else app_logger # Create sub-loggers for different components git_log = log.with_source("git") if context else log superset_log = log.with_source("superset_api") if context else log log.info(f"Executing operation: {operation} for dashboard {dashboard_id}") if operation == "sync": source_env_id = task_data.get("source_env_id") result = await self._handle_sync(dashboard_id, source_env_id, log, git_log, superset_log) elif operation == "deploy": env_id = task_data.get("environment_id") result = await self._handle_deploy(dashboard_id, env_id, log, git_log, superset_log) elif operation == "history": result = {"status": "success", "message": "History available via API"} else: log.error(f"Unknown operation: {operation}") raise ValueError(f"Unknown operation: {operation}") log.info(f"Operation {operation} completed.") return result # endregion execute # region _handle_sync [C:4] [TYPE Function] [SEMANTICS git,sync,backup,transactional] # @PURPOSE: Экспортирует дашборд из Superset и распаковывает в Git-репозиторий с backup/restore. # @PRE Репозиторий для дашборда должен существовать. # @POST Файлы в репозитории обновлены до текущего состояния в Superset. # Управляемые директории backup-ируются перед удалением; при ошибке распаковки backup восстанавливается. # @SIDE_EFFECT Изменяет файлы в локальной рабочей директории репозитория. # Создаёт временный backup в /tmp/ss-tools-backup-{dashboard_id}-{timestamp}/. # @RETURN Dict[str, str] - Результат синхронизации. # @RELATION CALLS -> src.services.git_service.GitService.get_repo # @RELATION CALLS -> src.core.superset_client.SupersetClient.export_dashboard async def _handle_sync(self, dashboard_id: int, source_env_id: str | None = None, log=None, git_log=None, superset_log=None) -> dict[str, str]: with belief_scope("GitPlugin._handle_sync"): try: repo = self.git_service.get_repo(dashboard_id) repo_path = Path(repo.working_dir) git_log.info(f"Target repo path: {repo_path}") env = self._get_env(source_env_id) client = SupersetClient(env) client.authenticate() superset_log.info(f"Exporting dashboard {dashboard_id} from {env.name}") zip_bytes, _ = client.export_dashboard(dashboard_id) git_log.info(f"Unpacking export to {repo_path}") managed_dirs = ["dashboards", "charts", "datasets", "databases"] managed_files = ["metadata.yaml"] # Fix 1: Create backup before deleting managed files backup_dir = Path(f"/tmp/ss-tools-backup-{dashboard_id}-{time.time_ns()}") for d in managed_dirs: d_path = repo_path / d if d_path.exists() and d_path.is_dir(): shutil.copytree(d_path, backup_dir / d) for f in managed_files: f_path = repo_path / f if f_path.exists(): (backup_dir / f).parent.mkdir(parents=True, exist_ok=True) shutil.copy2(f_path, backup_dir / f) # Delete old managed files for d in managed_dirs: d_path = repo_path / d if d_path.exists() and d_path.is_dir(): shutil.rmtree(d_path) for f in managed_files: f_path = repo_path / f if f_path.exists(): f_path.unlink() try: with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf: namelist = zf.namelist() if not namelist: raise ValueError("Export ZIP is empty") root_folder = namelist[0].split('/')[0] git_log.info(f"Detected root folder in ZIP: {root_folder}") for member in zf.infolist(): if member.filename.startswith(root_folder + "/") and len(member.filename) > len(root_folder) + 1: relative_path = member.filename[len(root_folder)+1:] target_path = repo_path / relative_path if member.is_dir(): target_path.mkdir(parents=True, exist_ok=True) else: target_path.parent.mkdir(parents=True, exist_ok=True) with zf.open(member) as source, open(target_path, "wb") as target: shutil.copyfileobj(source, target) except Exception: # Fix 1: Restore backup on unzip failure if backup_dir.exists(): for d in managed_dirs: src = backup_dir / d if src.exists(): dst = repo_path / d if dst.exists(): shutil.rmtree(dst) shutil.copytree(src, dst) for f in managed_files: src = backup_dir / f if src.exists(): dst = repo_path / f dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dst) raise # Fix 1: Remove backup on success if backup_dir.exists(): shutil.rmtree(backup_dir, ignore_errors=True) try: repo.git.add(A=True, force=True) app_logger.reason("Changes staged in git (force)", extra={"src": "_handle_sync"}) except Exception as ge: app_logger.explore(f"Failed to stage changes: {ge}", extra={"src": "_handle_sync"}) app_logger.info(f"[_handle_sync][Coherence:OK] Dashboard {dashboard_id} synced successfully.") return {"status": "success", "message": "Dashboard synced and flattened in local repository"} except Exception as e: app_logger.error(f"[_handle_sync][Coherence:Failed] Sync failed: {e}") raise # endregion _handle_sync # region _handle_deploy [C:4] [TYPE Function] [SEMANTICS git,deploy,zip] # @PURPOSE: Packages repository into ZIP and imports into target Superset environment. # @POST Dashboard imported into target Superset. # @SIDE_EFFECT Creates and removes temporary ZIP file. # @RELATION CALLS -> src.core.superset_client.SupersetClient.import_dashboard async def _handle_deploy(self, dashboard_id: int, env_id: str, log=None, git_log=None, superset_log=None) -> dict[str, Any]: with belief_scope("GitPlugin._handle_deploy"): try: if not env_id: raise ValueError("Target environment ID required for deployment") repo = self.git_service.get_repo(dashboard_id) repo_path = Path(repo.working_dir) git_log.info(f"Packing repository {repo_path} for deployment.") zip_buffer = io.BytesIO() root_dir_name = f"dashboard_export_{dashboard_id}" with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: for root, dirs, files in os.walk(repo_path): if ".git" in dirs: dirs.remove(".git") for file in files: if file == ".git" or file.endswith(".zip"): continue file_path = Path(root) / file arcname = Path(root_dir_name) / file_path.relative_to(repo_path) zf.write(file_path, arcname) zip_buffer.seek(0) env = self.config_manager.get_environment(env_id) if not env: raise ValueError(f"Environment {env_id} not found") client = SupersetClient(env) client.authenticate() temp_zip_path = repo_path / f"deploy_{dashboard_id}.zip" git_log.info(f"Saving temporary zip to {temp_zip_path}") with open(temp_zip_path, "wb") as f: f.write(zip_buffer.getvalue()) try: app_logger.reason(f"Importing dashboard to {env.name}", extra={"src": "_handle_deploy"}) result = client.import_dashboard(temp_zip_path) app_logger.info(f"[_handle_deploy][Coherence:OK] Deployment successful for dashboard {dashboard_id}.") return {"status": "success", "message": f"Dashboard deployed to {env.name}", "details": result} finally: if temp_zip_path.exists(): os.remove(temp_zip_path) except Exception as e: app_logger.error(f"[_handle_deploy][Coherence:Failed] Deployment failed: {e}") raise # endregion _handle_deploy # region _get_env [C:4] [TYPE Function] [SEMANTICS env,config,strict] # @PURPOSE: Вспомогательный метод для получения конфигурации окружения. # @PARAM env_id (Optional[str]) - ID окружения. # @PRE env_id is a string or None. # @POST Returns an Environment object from config or DB. # When env_id is explicitly provided and not found, raises ValueError (no fallback). # @RETURN Environment - Объект конфигурации окружения. def _get_env(self, env_id: str | None = None): with belief_scope("GitPlugin._get_env"): app_logger.reason(f"Fetching environment for ID: {env_id}", extra={"src": "_get_env"}) # Priority 1: ConfigManager (config.json) if env_id: env = self.config_manager.get_environment(env_id) if env: app_logger.reflect(f"Found environment by ID in ConfigManager: {env.name}", extra={"src": "_get_env"}) return env # Priority 2: Database (DeploymentEnvironment) from src.core.database import SessionLocal from src.models.git import DeploymentEnvironment db = SessionLocal() try: if env_id: db_env = db.query(DeploymentEnvironment).filter(DeploymentEnvironment.id == env_id).first() else: db_env = db.query(DeploymentEnvironment).filter(DeploymentEnvironment.is_active).first() if not db_env: db_env = db.query(DeploymentEnvironment).first() if db_env: app_logger.reflect(f"Found environment in DB: {db_env.name}", extra={"src": "_get_env"}) from src.core.config_models import Environment return Environment( id=db_env.id, name=db_env.name, url=db_env.superset_url, username="admin", password=db_env.superset_token, verify_ssl=True ) # Fix 7: Strict check — if env_id was explicitly provided but not found, raise immediately if env_id: raise ValueError(f"Environment '{env_id}' not found in ConfigManager or database") finally: db.close() # Priority 3: fallback to first env only when no specific env_id was requested envs = self.config_manager.get_environments() if envs and not env_id: app_logger.reflect(f"Using first environment from ConfigManager: {envs[0].name}", extra={"src": "_get_env"}) return envs[0] app_logger.error(f"[_get_env][Coherence:Failed] No environments configured (searched config.json and DB). env_id={env_id}") raise ValueError("No environments configured. Please add a Superset Environment in Settings.") # endregion _get_env # #endregion GitPlugin # #endregion GitPluginModule