# [MODULE] Superset Dashboard Migration Script # @contract: Автоматизирует процесс миграции и обновления дашбордов Superset между окружениями. # @semantic_layers: # 1. Конфигурация клиентов Superset для исходного и целевого окружений. # 2. Определение правил трансформации конфигураций баз данных. # 3. Экспорт дашборда, модификация YAML-файлов, создание нового архива и импорт. # @coherence: # - Использует `SupersetClient` для взаимодействия с API Superset. # - Использует `SupersetLogger` для централизованного логирования. # - Работает с `Pathlib` для управления файлами и директориями. # - Интегрируется с `keyring` для безопасного хранения паролей. # - Зависит от утилит `fileio` для обработки архивов и YAML-файлов. # [IMPORTS] Локальные модули from superset_tool.models import SupersetConfig from superset_tool.client import SupersetClient from superset_tool.utils.logger import SupersetLogger from superset_tool.exceptions import AuthenticationError, SupersetAPIError, NetworkError, DashboardNotFoundError from superset_tool.utils.fileio import save_and_unpack_dashboard, update_yamls, create_dashboard_export, create_temp_file, read_dashboard_from_disk from superset_tool.utils.init_clients import setup_clients # [IMPORTS] Стандартная библиотека import os import keyring from pathlib import Path import logging # [CONFIG] Инициализация глобального логгера # @invariant: Логгер доступен для всех компонентов скрипта. log_dir = Path("H:\\dev\\Logs") # [COHERENCE_NOTE] Убедитесь, что путь доступен. logger = SupersetLogger( log_dir=log_dir, level=logging.INFO, console=True ) logger.info("[COHERENCE_CHECK_PASSED] Логгер инициализирован для скрипта миграции.") # [CONFIG] Конфигурация трансформации базы данных Clickhouse # @semantic: Определяет, как UUID и URI базы данных Clickhouse должны быть изменены. # @invariant: 'old' и 'new' должны содержать полные конфигурации. database_config_click = { "old": { "database_name": "Prod Clickhouse", "sqlalchemy_uri": "clickhousedb+connect://clicketl:XXXXXXXXXX@rgm-s-khclk.hq.root.ad:443/dm", "uuid": "b9b67cb5-9874-4dc6-87bd-354fc33be6f9", "database_uuid": "b9b67cb5-9874-4dc6-87bd-354fc33be6f9", "allow_ctas": "false", "allow_cvas": "false", "allow_dml": "false" }, "new": { "database_name": "Dev Clickhouse", "sqlalchemy_uri": "clickhousedb+connect://dwhuser:XXXXXXXXXX@10.66.229.179:8123/dm", "uuid": "e9fd8feb-cb77-4e82-bc1d-44768b8d2fc2", "database_uuid": "e9fd8feb-cb77-4e82-bc1d-44768b8d2fc2", "allow_ctas": "true", "allow_cvas": "true", "allow_dml": "true" } } logger.debug("[CONFIG] Конфигурация Clickhouse загружена.") # [CONFIG] Конфигурация трансформации базы данных Greenplum # @semantic: Определяет, как UUID и URI базы данных Greenplum должны быть изменены. # @invariant: 'old' и 'new' должны содержать полные конфигурации. database_config_gp = { "old": { "database_name": "Prod Greenplum", "sqlalchemy_uri": "postgresql+psycopg2://viz_powerbi_gp_prod:XXXXXXXXXX@10.66.229.201:5432/dwh", "uuid": "805132a3-e942-40ce-99c7-bee8f82f8aa8", "database_uuid": "805132a3-e942-40ce-99c7-bee8f82f8aa8", "allow_ctas": "true", "allow_cvas": "true", "allow_dml": "true" }, "new": { "database_name": "DEV Greenplum", "sqlalchemy_uri": "postgresql+psycopg2://viz_superset_gp_dev:XXXXXXXXXX@10.66.229.171:5432/dwh", "uuid": "97b97481-43c3-4181-94c5-b69eaaa1e11f", "database_uuid": "97b97481-43c3-4181-94c5-b69eaaa1e11f", "allow_ctas": "false", "allow_cvas": "false", "allow_dml": "false" } } logger.debug("[CONFIG] Конфигурация Greenplum загружена.") # [ANCHOR] CLIENT_SETUP clients = setup_clients(logger) # [CONFIG] Определение исходного и целевого клиентов для миграции # [COHERENCE_NOTE] Эти переменные задают конкретную миграцию. Для параметризации можно использовать аргументы командной строки. from_c = clients["sbx"] # Источник миграции to_c = clients["preprod"] # Цель миграции dashboard_slug = "FI0060" # Идентификатор дашборда для миграции # dashboard_id = 53 # ID не нужен, если есть slug # [CONTRACT] # Описание: Мигрирует один дашборд с from_c на to_c. # @pre: # - from_c и to_c должны быть инициализированы. # @post: # - Дашборд с from_c успешно экспортирован и импортирован в to_c. # @raise: # - Exception: В случае ошибки экспорта или импорта. def migrate_dashboard (dashboard_slug=dashboard_slug, from_c = from_c, to_c = to_c, logger=logger, update_db_yaml=False): logger.info(f"[INFO] Конфигурация миграции: From '{from_c.config.base_url}' To '{to_c.config.base_url}' for dashboard slug '{dashboard_slug}'") try: # [ACTION] Получение метаданных исходного дашборда logger.info(f"[INFO] Получение метаданных дашборда '{dashboard_slug}' из исходного окружения.") dashboard_meta = from_c.get_dashboard(dashboard_slug) dashboard_id = dashboard_meta["id"] # Получаем ID из метаданных logger.info(f"[INFO] Найден дашборд '{dashboard_meta['dashboard_title']}' с ID: {dashboard_id}.") # [CONTEXT_MANAGER] Работа с временной директорией для обработки архива дашборда with create_temp_file(suffix='.dir', logger=logger) as temp_root: logger.info(f"[INFO] Создана временная директория: {temp_root}") # [ANCHOR] EXPORT_DASHBOARD # Экспорт дашборда во временную директорию ИЛИ чтение с диска # [COHERENCE_NOTE] В текущем коде закомментирован экспорт и используется локальный файл. # Для полноценной миграции следует использовать export_dashboard(). zip_content, filename = from_c.export_dashboard(dashboard_id) # Предпочтительный путь для реальной миграции # [DEBUG] Использование файла с диска для тестирования миграции #zip_db_path = r"C:\Users\VolobuevAA\Downloads\dashboard_export_20250704T082538.zip" #logger.warning(f"[WARN] Используется ЛОКАЛЬНЫЙ файл дашборда для миграции: {zip_db_path}. Это может привести к некогерентности, если файл устарел.") #zip_content, filename = read_dashboard_from_disk(zip_db_path, logger=logger) # [ANCHOR] SAVE_AND_UNPACK # Сохранение и распаковка во временную директорию zip_path, unpacked_path = save_and_unpack_dashboard( zip_content=zip_content, original_filename=filename, unpack=True, logger=logger, output_dir=temp_root ) logger.info(f"[INFO] Дашборд распакован во временную директорию: {unpacked_path}") # [ANCHOR] UPDATE_YAML_CONFIGS # Обновление конфигураций баз данных в YAML-файлах if update_db_yaml: source_path = unpacked_path / Path(filename).stem # Путь к распакованному содержимому дашборда db_configs_to_apply = [database_config_click, database_config_gp] logger.info(f"[INFO] Применение трансформаций баз данных к YAML файлам в {source_path}...") update_yamls(db_configs_to_apply, path=source_path, logger=logger) logger.info("[INFO] YAML-файлы успешно обновлены.") # [ANCHOR] CREATE_NEW_EXPORT_ARCHIVE # Создание нового экспорта дашборда из модифицированных файлов temp_zip = temp_root / f"{dashboard_slug}_migrated.zip" # Имя файла для импорта logger.info(f"[INFO] Создание нового ZIP-архива для импорта: {temp_zip}") create_dashboard_export(temp_zip, [source_path], logger=logger) logger.info("[INFO] Новый ZIP-архив дашборда готов к импорту.") else: temp_zip = zip_path # [ANCHOR] IMPORT_DASHBOARD # Импорт обновленного дашборда в целевое окружение logger.info(f"[INFO] Запуск импорта дашборда в целевое окружение {to_c.config.base_url}...") import_result = to_c.import_dashboard(temp_zip) logger.info(f"[COHERENCE_CHECK_PASSED] Дашборд '{dashboard_slug}' успешно импортирован/обновлен.", extra={"import_result": import_result}) except (AuthenticationError, SupersetAPIError, NetworkError, DashboardNotFoundError) as e: logger.error(f"[ERROR] Ошибка миграции дашборда: {str(e)}", exc_info=True, extra=e.context) # exit(1) except Exception as e: logger.critical(f"[CRITICAL] Фатальная и необработанная ошибка в скрипте миграции: {str(e)}", exc_info=True) # exit(1) logger.info("[INFO] Процесс миграции завершен.") # [CONTRACT] # Описание: Мигрирует все дашборды с from_c на to_c. # @pre: # - from_c и to_c должны быть инициализированы. # @post: # - Все дашборды с from_c успешно экспортированы и импортированы в to_c. # @raise: # - Exception: В случае ошибки экспорта или импорта. def migrate_all_dashboards(from_c: SupersetClient, to_c: SupersetClient,logger=logger) -> None: # [ACTION] Получение списка всех дашбордов из исходного окружения. logger.info(f"[ACTION] Получение списка всех дашбордов из '{from_c.config.base_url}'") total_dashboards, dashboards = from_c.get_dashboards() logger.info(f"[INFO] Найдено {total_dashboards} дашбордов для миграции.") # [ACTION] Итерация по всем дашбордам и миграция каждого из них. for dashboard in dashboards: dashboard_id = dashboard["id"] dashboard_slug = dashboard["slug"] dashboard_title = dashboard["dashboard_title"] logger.info(f"[INFO] Начало миграции дашборда '{dashboard_title}' (ID: {dashboard_id}, Slug: {dashboard_slug}).") if dashboard_slug: try: migrate_dashboard(dashboard_slug=dashboard_slug,from_c=from_c,to_c=to_c,logger=logger) except Exception as e: logger.error(f"[ERROR] Ошибка миграции дашборда: {str(e)}", exc_info=True, extra=e.context) else: logger.info(f"[INFO] Пропуск '{dashboard_title}' (ID: {dashboard_id}, Slug: {dashboard_slug}). Пустой SLUG") logger.info(f"[INFO] Миграция всех дашбордов с '{from_c.config.base_url}' на '{to_c.config.base_url}' завершена.") # [ACTION] Вызов функции миграции migrate_all_dashboards(from_c, to_c)