diff --git a/backend/src/plugins/backup.py b/backend/src/plugins/backup.py index 3d70d151..2e218c0b 100755 --- a/backend/src/plugins/backup.py +++ b/backend/src/plugins/backup.py @@ -15,6 +15,7 @@ from ..core.logger import belief_scope, logger as app_logger from ..core.plugin_base import PluginBase from ..core.superset_client import SupersetClient from ..core.task_manager.context import TaskContext +from ..core.utils.executors import run_blocking from ..core.utils.fileio import RetentionPolicy, archive_exports, consolidate_archive_folders, remove_empty_directories, sanitize_filename, save_and_unpack_dashboard from ..core.utils.network import SupersetAPIError from ..dependencies import get_config_manager @@ -159,8 +160,8 @@ class BackupPlugin(PluginBase): client = SupersetClient(env_config) - # Get all dashboards - all_dashboard_count, all_dashboard_meta = client.get_dashboards() + # Get all dashboards (async) + all_dashboard_count, all_dashboard_meta = await client.get_dashboards() superset_log.info(f"Found {all_dashboard_count} total dashboards in environment") # Filter dashboards if specific IDs are provided @@ -198,24 +199,31 @@ class BackupPlugin(PluginBase): # Report progress progress_pct = (idx / total) * 100 - log.progress(f"Backing up dashboard: {dashboard_title}", percent=progress_pct) + log.progress(f"Backing up: {dashboard_title}", percent=progress_pct) try: dashboard_base_dir_name = sanitize_filename(f"{dashboard_title}") dashboard_dir = backup_path / env.upper() / dashboard_base_dir_name - dashboard_dir.mkdir(parents=True, exist_ok=True) + await run_blocking(kind='file', fn=dashboard_dir.mkdir, parents=True, exist_ok=True) - zip_content, filename = client.export_dashboard(dashboard_id) + zip_content, filename = await client.export_dashboard(dashboard_id) superset_log.debug(f"Exported dashboard: {dashboard_title}") - save_and_unpack_dashboard( + await run_blocking( + kind='file', + fn=save_and_unpack_dashboard, zip_content=zip_content, original_filename=filename, output_dir=dashboard_dir, - unpack=False + unpack=False, ) - archive_exports(str(dashboard_dir), policy=RetentionPolicy()) + await run_blocking( + kind='file', + fn=archive_exports, + output_dir=str(dashboard_dir), + policy=RetentionPolicy(), + ) storage_log.debug(f"Archived dashboard: {dashboard_title}") backed_up_dashboards.append({ "id": dashboard_id, @@ -232,8 +240,16 @@ class BackupPlugin(PluginBase): }) continue - consolidate_archive_folders(backup_path / env.upper()) - remove_empty_directories(str(backup_path / env.upper())) + await run_blocking( + kind='file', + fn=consolidate_archive_folders, + root_directory=backup_path / env.upper(), + ) + await run_blocking( + kind='file', + fn=remove_empty_directories, + root_dir=str(backup_path / env.upper()), + ) log.info(f"Backup completed successfully for {env}") return { diff --git a/backend/src/plugins/git_plugin.py b/backend/src/plugins/git_plugin.py index 9963a36a..b5620c55 100644 --- a/backend/src/plugins/git_plugin.py +++ b/backend/src/plugins/git_plugin.py @@ -26,9 +26,102 @@ 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.core.utils.executors import run_blocking from src.services.git_service import GitService +# #region _sync_helpers [C:2] [TYPE Module] [SEMANTICS git,sync,helpers] +# @BRIEF Helper functions for GitPlugin._handle_sync, extracted for use with run_blocking. +# @RELATION DEPENDS_ON -> [EXT:shutil] +# @RELATION DEPENDS_ON -> [EXT:zipfile] + + +def _create_sync_backup(repo_path: Path, backup_dir: Path, managed_dirs: list[str], managed_files: list[str]) -> None: + """Backup managed directories and files before sync.""" + 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) + + +def _delete_managed_files(repo_path: Path, managed_dirs: list[str], managed_files: list[str]) -> None: + """Delete old managed files from repo before unpacking.""" + 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() + + +def _extract_zip_to_repo(zip_bytes: bytes, repo_path: Path) -> None: + """Extract dashboard export ZIP into the repository directory.""" + 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] + 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) + + +def _restore_sync_backup(repo_path: Path, backup_dir: Path, managed_dirs: list[str], managed_files: list[str]) -> None: + """Restore managed files from backup after a failed sync.""" + 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) + + +# #endregion _sync_helpers + + +# #region _deploy_helpers [C:2] [TYPE Module] [SEMANTICS git,deploy,zip] +# @BRIEF Helper functions for GitPlugin._handle_deploy, extracted for use with run_blocking. + + +def _pack_deploy_zip(repo_path: Path, root_dir_name: str, zip_buffer: io.BytesIO) -> None: + """Walk repo_path and pack files into zip_buffer (thread-safe).""" + 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) + + +# #endregion _deploy_helpers + + # #region GitPlugin [TYPE Class] # @BRIEF Реализация плагина Git Integration для управления версиями дашбордов. class GitPlugin(PluginBase): @@ -154,85 +247,49 @@ class GitPlugin(PluginBase): 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 = await run_blocking(kind='git', fn=self.git_service.get_repo, dashboard_id=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() + await client.authenticate() superset_log.info(f"Exporting dashboard {dashboard_id} from {env.name}") - zip_bytes, _ = client.export_dashboard(dashboard_id) + zip_bytes, _ = await 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 + # Fix 1: Create backup before deleting managed files (wrapped in run_blocking) 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) + + await run_blocking(kind='file', fn=_create_sync_backup, repo_path=repo_path, backup_dir=backup_dir, managed_dirs=managed_dirs, managed_files=managed_files) # 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() + await run_blocking(kind='file', fn=_delete_managed_files, repo_path=repo_path, managed_dirs=managed_dirs, managed_files=managed_files) 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) + await run_blocking( + kind='file', + fn=_extract_zip_to_repo, + zip_bytes=zip_bytes, + repo_path=repo_path, + ) 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) + await run_blocking(kind='file', fn=_restore_sync_backup, repo_path=repo_path, backup_dir=backup_dir, managed_dirs=managed_dirs, managed_files=managed_files) raise # Fix 1: Remove backup on success if backup_dir.exists(): - shutil.rmtree(backup_dir, ignore_errors=True) + await run_blocking(kind='file', fn=shutil.rmtree, path=backup_dir, ignore_errors=True) try: - repo.git.add(A=True, force=True) + await run_blocking(kind='git', fn=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"}) @@ -255,40 +312,43 @@ class GitPlugin(PluginBase): try: if not env_id: raise ValueError("Target environment ID required for deployment") - repo = self.git_service.get_repo(dashboard_id) + repo = await run_blocking(kind='git', fn=self.git_service.get_repo, dashboard_id=dashboard_id) repo_path = Path(repo.working_dir) git_log.info(f"Packing repository {repo_path} for deployment.") + + # Build ZIP in thread pool to avoid blocking event loop with os.walk 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) + await run_blocking( + kind='file', + fn=_pack_deploy_zip, + repo_path=repo_path, + root_dir_name=root_dir_name, + zip_buffer=zip_buffer, + ) + env = self.config_manager.get_environment(env_id) if not env: raise ValueError(f"Environment {env_id} not found") client = SupersetClient(env) - client.authenticate() + await 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()) + await run_blocking( + kind='file', + fn=temp_zip_path.write_bytes, + data=zip_buffer.getvalue(), + ) try: app_logger.reason(f"Importing dashboard to {env.name}", extra={"src": "_handle_deploy"}) - result = client.import_dashboard(temp_zip_path) + result = await 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) + if await run_blocking(kind='file', fn=temp_zip_path.exists): + await run_blocking(kind='file', fn=os.remove, path=temp_zip_path) except Exception as e: app_logger.error(f"[_handle_deploy][Coherence:Failed] Deployment failed: {e}") diff --git a/backend/src/plugins/llm_analysis/plugin.py b/backend/src/plugins/llm_analysis/plugin.py index 0b5b0124..c694da83 100644 --- a/backend/src/plugins/llm_analysis/plugin.py +++ b/backend/src/plugins/llm_analysis/plugin.py @@ -602,7 +602,7 @@ class DashboardValidationPlugin(PluginBase): # 1. Fetch dashboard metadata from Superset superset_log.info(f"Fetching dashboard {dashboard_id} metadata for Path B") client = SupersetClient(env) - dashboard_raw = client.get_dashboard(dashboard_id) + dashboard_raw = await client.get_dashboard(dashboard_id) dashboard = dashboard_raw.get("result", dashboard_raw) if isinstance(dashboard_raw, dict) else dashboard_raw slices = dashboard.get("slices", []) @@ -613,7 +613,7 @@ class DashboardValidationPlugin(PluginBase): slice_id = slice_ref.get("id") or slice_ref.get("slice_id") if slice_id: try: - chart_raw = client.get_chart(int(slice_id)) + chart_raw = await client.get_chart(int(slice_id)) chart = chart_raw.get("result", chart_raw) if isinstance(chart_raw, dict) else chart_raw chart_data.append(chart) chart_entries.append({ @@ -808,7 +808,7 @@ class DashboardValidationPlugin(PluginBase): } superset_log.debug(f"Fetching logs for dashboard {dashboard_id}") - response = client.network.request( + response = await client.network.request( method="GET", endpoint="/log/", params={"q": json.dumps(query_params)}, @@ -937,7 +937,7 @@ class DocumentationPlugin(PluginBase): client = SupersetClient(env) superset_log.debug(f"Fetching dataset {dataset_id}") - dataset = client.get_dataset(int(dataset_id)) + dataset = await client.get_dataset(int(dataset_id)) # Extract columns and existing descriptions columns_data = [] @@ -990,7 +990,7 @@ class DocumentationPlugin(PluginBase): }) superset_log.info(f"Updating dataset {dataset_id} with generated documentation") - client.update_dataset(int(dataset_id), update_payload) + await client.update_dataset(int(dataset_id), update_payload) log.info(f"Documentation completed for dataset {dataset_id}") diff --git a/backend/src/plugins/storage/plugin.py b/backend/src/plugins/storage/plugin.py index 355940b6..be559356 100644 --- a/backend/src/plugins/storage/plugin.py +++ b/backend/src/plugins/storage/plugin.py @@ -18,10 +18,21 @@ from fastapi import UploadFile from ...core.logger import belief_scope, logger from ...core.plugin_base import PluginBase from ...core.task_manager.context import TaskContext +from ...core.utils.executors import run_blocking from ...dependencies import get_config_manager from ...models.storage import FileCategory, StoredFile +# #region _copy_upload_to_disk [C:2] [TYPE Function] +# @BRIEF Helper for save_file — writes uploaded file content to disk synchronously in a thread pool. +# @PRE dest_path is a validated path. file_obj is an open SpooledTemporaryFile. +# @POST File content written to disk at dest_path. +def _copy_upload_to_disk(dest_path: Path, file_obj) -> None: + with open(dest_path, "wb") as buffer: + shutil.copyfileobj(file_obj, buffer) +# #endregion _copy_upload_to_disk + + # #region StoragePlugin [TYPE Class] # @BRIEF Implementation of the storage management plugin. class StoragePlugin(PluginBase): @@ -321,12 +332,12 @@ class StoragePlugin(PluginBase): if subpath: dest_dir = dest_dir / subpath - dest_dir.mkdir(parents=True, exist_ok=True) + await run_blocking(kind='file', fn=dest_dir.mkdir, parents=True, exist_ok=True) dest_path = self.validate_path(dest_dir / file.filename) - with dest_path.open("wb") as buffer: - shutil.copyfileobj(file.file, buffer) + # Write file content via run_blocking to avoid blocking the event loop + await run_blocking(kind='file', fn=_copy_upload_to_disk, dest_path=dest_path, file_obj=file.file) stat = dest_path.stat() return StoredFile( @@ -346,7 +357,7 @@ class StoragePlugin(PluginBase): # @PRE path must belong to the specified category and exist on disk. # @POST The file or directory is removed from disk. # @SIDE_EFFECT Removes item from disk. - def delete_file(self, category: FileCategory, path: str): + async def delete_file(self, category: FileCategory, path: str): with belief_scope("StoragePlugin:delete_file"): root = self.get_storage_root() # path is relative to root, but we ensure it starts with category @@ -357,9 +368,9 @@ class StoragePlugin(PluginBase): if full_path.exists(): if full_path.is_dir(): - shutil.rmtree(full_path) + await run_blocking(kind='file', fn=shutil.rmtree, path=full_path) else: - full_path.unlink() + await run_blocking(kind='file', fn=full_path.unlink) logger.reason(f"Deleted: {full_path}", extra={"src": "StoragePlugin.delete_file"}) else: raise FileNotFoundError(f"Item {path} not found")