Merge branch 'gemini-refactor' into 'master'
add clean folders in backup See merge request dwh_bi/superset-tools!2
This commit is contained in:
Волобуев Андрей Александрович (VolobuevAA)
@@ -24,7 +24,7 @@ import keyring
|
||||
from superset_tool.models import SupersetConfig
|
||||
from superset_tool.client import SupersetClient
|
||||
from superset_tool.utils.logger import SupersetLogger
|
||||
from superset_tool.utils.fileio import save_and_unpack_dashboard, archive_exports, sanitize_filename
|
||||
from superset_tool.utils.fileio import save_and_unpack_dashboard, archive_exports, sanitize_filename,consolidate_archive_folders,remove_empty_directories
|
||||
# [COHERENCE_CHECK_PASSED] Все необходимые модули импортированы и согласованы.
|
||||
|
||||
# [FUNCTION] setup_clients
|
||||
@@ -92,25 +92,35 @@ def setup_clients(logger: SupersetLogger):
|
||||
raise
|
||||
|
||||
# [FUNCTION] backup_dashboards
|
||||
# @contract: Выполняет бэкап всех доступных дашбордов для заданного клиента и окружения.
|
||||
# @pre:
|
||||
# - `client` должен быть инициализированным экземпляром `SupersetClient`.
|
||||
# - `env_name` должен быть строкой, обозначающей окружение.
|
||||
# - `backup_root` должен быть валидным путем к корневой директории бэкапа.
|
||||
# - `logger` должен быть инициализирован.
|
||||
# @post:
|
||||
# - Дашборды экспортируются и сохраняются в поддиректориях `backup_root/env_name/dashboard_title`.
|
||||
# - Старые экспорты архивируются.
|
||||
# - Возвращает `True` если все дашборды были экспортированы без критических ошибок, `False` иначе.
|
||||
# @side_effects:
|
||||
# - Создает директории и файлы в файловой системе.
|
||||
# - Логирует статус выполнения, успешные экспорты и ошибки.
|
||||
# @exceptions:
|
||||
# - `SupersetAPIError`, `NetworkError`, `DashboardNotFoundError`, `ExportError` могут быть подняты методами `SupersetClient` и будут логированы.
|
||||
def backup_dashboards(client: SupersetClient, env_name: str, backup_root: Path, logger: SupersetLogger) -> bool:
|
||||
"""Выполнение бэкапа дашбордов с детальным логированием ошибок"""
|
||||
def backup_dashboards(client: SupersetClient,
|
||||
env_name: str,
|
||||
backup_root: Path,
|
||||
logger: SupersetLogger,
|
||||
consolidate: bool = True,
|
||||
rotate_archive: bool = True,
|
||||
clean_folders:bool = True) -> bool:
|
||||
""" [CONTRACT] Выполняет бэкап всех доступных дашбордов для заданного клиента и окружения.
|
||||
@pre:
|
||||
- `client` должен быть инициализированным экземпляром `SupersetClient`.
|
||||
- `env_name` должен быть строкой, обозначающей окружение.
|
||||
- `backup_root` должен быть валидным путем к корневой директории бэкапа.
|
||||
- `logger` должен быть инициализирован.
|
||||
@post:
|
||||
- Дашборды экспортируются и сохраняются в поддиректориях `backup_root/env_name/dashboard_title`.
|
||||
- Старые экспорты архивируются.
|
||||
- Возвращает `True` если все дашборды были экспортированы без критических ошибок, `False` иначе.
|
||||
@side_effects:
|
||||
- Создает директории и файлы в файловой системе.
|
||||
- Логирует статус выполнения, успешные экспорты и ошибки.
|
||||
@exceptions:
|
||||
- `SupersetAPIError`, `NetworkError`, `DashboardNotFoundError`, `ExportError` могут быть подняты методами `SupersetClient` и будут логированы."""
|
||||
# [ANCHOR] DASHBOARD_BACKUP_PROCESS
|
||||
logger.info(f"[INFO] Запуск бэкапа дашбордов для окружения: {env_name}")
|
||||
logger.debug(
|
||||
"[PARAMS] Флаги: consolidate=%s, rotate_archive=%s, clean_folders=%s",
|
||||
consolidate, rotate_archive, clean_folders,
|
||||
extra={"env": env_name} # контекст для логирования
|
||||
)
|
||||
try:
|
||||
dashboard_count, dashboard_meta = client.get_dashboards()
|
||||
logger.info(f"[INFO] Найдено {dashboard_count} дашбордов для экспорта в {env_name}")
|
||||
@@ -139,7 +149,7 @@ def backup_dashboards(client: SupersetClient, env_name: str, backup_root: Path,
|
||||
try:
|
||||
# [ANCHOR] CREATE_DASHBOARD_DIR
|
||||
# Используем slug в пути для большей уникальности и избежания конфликтов имен
|
||||
dashboard_base_dir_name = sanitize_filename(f"{dashboard_slug}-{dashboard_title}")
|
||||
dashboard_base_dir_name = sanitize_filename(f"{dashboard_title}")
|
||||
dashboard_dir = backup_root / env_name / dashboard_base_dir_name
|
||||
dashboard_dir.mkdir(parents=True, exist_ok=True)
|
||||
logger.debug(f"[DEBUG] Директория для дашборда: {dashboard_dir}")
|
||||
@@ -158,18 +168,20 @@ def backup_dashboards(client: SupersetClient, env_name: str, backup_root: Path,
|
||||
)
|
||||
logger.info(f"[INFO] Дашборд '{dashboard_title}' (ID: {dashboard_id}) успешно экспортирован.")
|
||||
|
||||
if rotate_archive:
|
||||
# [ANCHOR] ARCHIVE_OLD_BACKUPS
|
||||
try:
|
||||
archive_exports(dashboard_dir, logger=logger)
|
||||
logger.debug(f"[DEBUG] Старые экспорты для '{dashboard_title}' архивированы.")
|
||||
except Exception as cleanup_error:
|
||||
logger.warning(
|
||||
f"[WARN] Ошибка архивирования старых бэкапов для '{dashboard_title}': {cleanup_error}",
|
||||
exc_info=False # Не показываем полный traceback для очистки, т.к. это второстепенно
|
||||
)
|
||||
try:
|
||||
archive_exports(dashboard_dir, logger=logger)
|
||||
logger.debug(f"[DEBUG] Старые экспорты для '{dashboard_title}' архивированы.")
|
||||
except Exception as cleanup_error:
|
||||
logger.warning(
|
||||
f"[WARN] Ошибка архивирования старых бэкапов для '{dashboard_title}': {cleanup_error}",
|
||||
exc_info=False # Не показываем полный traceback для очистки, т.к. это второстепенно
|
||||
)
|
||||
|
||||
success_count += 1
|
||||
|
||||
|
||||
except Exception as db_error:
|
||||
error_info = {
|
||||
'dashboard_id': dashboard_id,
|
||||
@@ -184,6 +196,28 @@ def backup_dashboards(client: SupersetClient, env_name: str, backup_root: Path,
|
||||
extra=error_info, exc_info=True # Логируем полный traceback для ошибок экспорта
|
||||
)
|
||||
|
||||
if consolidate:
|
||||
# [ANCHOR] Объединяем архивы по SLUG в одну папку с максимальной датой
|
||||
try:
|
||||
consolidate_archive_folders(backup_root / env_name , logger=logger)
|
||||
logger.debug(f"[DEBUG] Файлы для '{dashboard_title}' консолидированы.")
|
||||
except Exception as consolidate_error:
|
||||
logger.warning(
|
||||
f"[WARN] Ошибка консолидации файлов для '{backup_root / env_name}': {consolidate_error}",
|
||||
exc_info=False # Не показываем полный traceback для консолидации, т.к. это второстепенно
|
||||
)
|
||||
|
||||
if clean_folders:
|
||||
# [ANCHOR] Удаляем пустые папки
|
||||
try:
|
||||
dirs_count = remove_empty_directories(backup_root / env_name , logger=logger)
|
||||
logger.debug(f"[DEBUG] {dirs_count} пустых папок в '{backup_root / env_name }' удалены.")
|
||||
except Exception as clean_error:
|
||||
logger.warning(
|
||||
f"[WARN] Ошибка очистки пустых директорий в '{backup_root / env_name}': {clean_error}",
|
||||
exc_info=False # Не показываем полный traceback для консолидации, т.к. это второстепенно
|
||||
)
|
||||
|
||||
if error_details:
|
||||
logger.error(
|
||||
f"[COHERENCE_CHECK_FAILED] Итоги экспорта для {env_name}:",
|
||||
|
||||
@@ -16,6 +16,7 @@ from pathlib import Path
|
||||
from typing import Any, Optional, Tuple, Dict, List, Literal, Union, BinaryIO, LiteralString
|
||||
from collections import defaultdict
|
||||
from datetime import date
|
||||
import glob
|
||||
from contextlib import contextmanager
|
||||
|
||||
# [IMPORTS] Third-party
|
||||
@@ -674,6 +675,85 @@ def update_yamls(
|
||||
logger.error(f"[YAML_UPDATE_ERROR] Критическая ошибка: {str(e)}", exc_info=True)
|
||||
raise
|
||||
|
||||
def consolidate_archive_folders(root_directory: Path, logger: Optional[SupersetLogger] = None) -> None:
|
||||
"""
|
||||
Consolidates dashboard folders under a root directory based on the slug (pattern MM-0080 - two latin letters - hyphen - 4 digits)
|
||||
and moves the contents to the folder with the latest modification date.
|
||||
|
||||
Args:
|
||||
root_directory (Path): The root directory containing the dashboard folders.
|
||||
|
||||
Raises:
|
||||
TypeError: If root_directory is not a Path object.
|
||||
ValueError: If root_directory is empty.
|
||||
|
||||
[CONTRACT]
|
||||
@pre: root_directory must be a valid Path object representing an existing directory.
|
||||
@post: The contents of all folders matching the slug pattern are moved to the folder with the latest modification date for each slug.
|
||||
@invariant: The slug pattern remains consistent throughout the execution.
|
||||
@raise: TypeError if root_directory is not a Path, ValueError if root_directory is empty.
|
||||
"""
|
||||
|
||||
# [CONTRACT] Ensure valid input
|
||||
if not isinstance(root_directory, Path):
|
||||
raise TypeError("root_directory must be a Path object.")
|
||||
if not root_directory:
|
||||
raise ValueError("root_directory cannot be empty.")
|
||||
|
||||
logger.debug(f"[DEBUG] Checking root_folder: {root_directory}")
|
||||
|
||||
# [SECTION] Define the slug pattern
|
||||
slug_pattern = re.compile(r"([A-Z]{2}-\d{4})") # Capture the first occurrence of the pattern
|
||||
|
||||
# [SECTION] Group folders by slug
|
||||
dashboards_by_slug: dict[str, list[str]] = {}
|
||||
for folder_name in glob.glob(os.path.join(root_directory, '*')):
|
||||
if os.path.isdir(folder_name):
|
||||
logger.debug(f"[DEBUG] Checking folder: {folder_name}") # Debug log: show the folder name being checked
|
||||
match = slug_pattern.search(folder_name)
|
||||
if match:
|
||||
slug = match.group(1) # Extract the captured group (the slug)
|
||||
logger.info(f"[INFO] Found slug: {slug} in folder: {folder_name}") #Log when slug is matched
|
||||
logger.debug(f"[DEBUG] Regex match object: {match}") # Log the complete match object
|
||||
if slug not in dashboards_by_slug:
|
||||
dashboards_by_slug[slug] = []
|
||||
dashboards_by_slug[slug].append(folder_name)
|
||||
else:
|
||||
logger.debug(f"[DEBUG] No slug found in folder: {folder_name}") # Debug log: show when slug is not matched
|
||||
else:
|
||||
logger.debug(f"[DEBUG] Not a directory: {folder_name}") #debug log for when its not a directory
|
||||
|
||||
# [SECTION] Check if any slugs were found
|
||||
if not dashboards_by_slug:
|
||||
logger.warning("[WARN] No folders found matching the slug pattern.")
|
||||
return
|
||||
|
||||
# [SECTION] Iterate through each slug group
|
||||
for slug, folder_list in dashboards_by_slug.items():
|
||||
# [ACTION] Find the folder with the latest modification date
|
||||
latest_folder = max(folder_list, key=os.path.getmtime)
|
||||
logger.info(f"[INFO] Latest folder for slug {slug}: {latest_folder}")
|
||||
|
||||
# [SECTION] Move contents of other folders to the latest folder
|
||||
for folder in folder_list:
|
||||
if folder != latest_folder:
|
||||
# [ACTION] Move contents
|
||||
try:
|
||||
for item in os.listdir(folder):
|
||||
s = os.path.join(folder, item)
|
||||
d = os.path.join(latest_folder, item)
|
||||
if os.path.isdir(s):
|
||||
shutil.move(s, d)
|
||||
else:
|
||||
shutil.move(s, d)
|
||||
|
||||
logger.info(f"[INFO] Moved contents of {folder} to {latest_folder}")
|
||||
except Exception as e:
|
||||
logger.error(f"[ERROR] Failed to move contents of {folder} to {latest_folder}: {e}", exc_info=True)
|
||||
|
||||
logger.info("[INFO] Dashboard consolidation completed.")
|
||||
# [COHERENCE_CHECK_PASSED] Function executed successfully and all contracts were met.
|
||||
|
||||
def sync_for_git(
|
||||
source_path: str,
|
||||
destination_path: str,
|
||||
|
||||
Reference in New Issue
Block a user