032: Phase 5 US3 — backup, git, llm_analysis, storage async

T032-T035 completed. T036 partial.
Remaining: T036 superset_compilation_adapter, T037 fileio.py, cascade updates

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 {

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}")

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}")
 

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")