refactor, add db search
This commit is contained in:
@@ -1,37 +1,38 @@
|
||||
# <GRACE_MODULE id="dataset_mapper" name="dataset_mapper.py">
|
||||
# @SEMANTICS: dataset, mapping, postgresql, xlsx, superset
|
||||
# @PURPOSE: Этот модуль отвечает за обновление метаданных (verbose_map) в датасетах Superset, извлекая их из PostgreSQL или XLSX-файлов.
|
||||
# @DEPENDS_ON: superset_tool.client -> Использует SupersetClient для взаимодействия с API.
|
||||
# @DEPENDS_ON: pandas -> для чтения XLSX-файлов.
|
||||
# @DEPENDS_ON: psycopg2 -> для подключения к PostgreSQL.
|
||||
|
||||
# <IMPORTS>
|
||||
import pandas as pd
|
||||
import psycopg2
|
||||
# [DEF:superset_tool.utils.dataset_mapper:Module]
|
||||
#
|
||||
# @SEMANTICS: dataset, mapping, postgresql, xlsx, superset
|
||||
# @PURPOSE: Этот модуль отвечает за обновление метаданных (verbose_map) в датасетах Superset, извлекая их из PostgreSQL или XLSX-файлов.
|
||||
# @LAYER: Domain
|
||||
# @RELATION: DEPENDS_ON -> superset_tool.client
|
||||
# @RELATION: DEPENDS_ON -> pandas
|
||||
# @RELATION: DEPENDS_ON -> psycopg2
|
||||
# @PUBLIC_API: DatasetMapper
|
||||
|
||||
# [SECTION: IMPORTS]
|
||||
import pandas as pd # type: ignore
|
||||
import psycopg2 # type: ignore
|
||||
from superset_tool.client import SupersetClient
|
||||
from superset_tool.utils.init_clients import setup_clients
|
||||
from superset_tool.utils.logger import SupersetLogger
|
||||
from typing import Dict, List, Optional, Any
|
||||
# </IMPORTS>
|
||||
|
||||
# --- Начало кода модуля ---
|
||||
|
||||
# <ANCHOR id="DatasetMapper" type="Class">
|
||||
# @PURPOSE: Класс для меппинга и обновления verbose_map в датасетах Superset.
|
||||
# [/SECTION]
|
||||
|
||||
# [DEF:DatasetMapper:Class]
|
||||
# @PURPOSE: Класс для меппинга и обновления verbose_map в датасетах Superset.
|
||||
class DatasetMapper:
|
||||
def __init__(self, logger: SupersetLogger):
|
||||
self.logger = logger
|
||||
|
||||
# <ANCHOR id="DatasetMapper.get_postgres_comments" type="Function">
|
||||
# @PURPOSE: Извлекает комментарии к колонкам из системного каталога PostgreSQL.
|
||||
# @PRE: `db_config` должен содержать валидные креды для подключения к PostgreSQL.
|
||||
# @PRE: `table_name` и `table_schema` должны быть строками.
|
||||
# @POST: Возвращается словарь с меппингом `column_name` -> `column_comment`.
|
||||
# @PARAM: db_config: Dict - Конфигурация для подключения к БД.
|
||||
# @PARAM: table_name: str - Имя таблицы.
|
||||
# @PARAM: table_schema: str - Схема таблицы.
|
||||
# @RETURN: Dict[str, str] - Словарь с комментариями к колонкам.
|
||||
# @THROW: Exception - При ошибках подключения или выполнения запроса к БД.
|
||||
|
||||
# [DEF:DatasetMapper.get_postgres_comments:Function]
|
||||
# @PURPOSE: Извлекает комментарии к колонкам из системного каталога PostgreSQL.
|
||||
# @PRE: `db_config` должен содержать валидные креды для подключения к PostgreSQL.
|
||||
# @PRE: `table_name` и `table_schema` должны быть строками.
|
||||
# @POST: Возвращается словарь с меппингом `column_name` -> `column_comment`.
|
||||
# @THROW: Exception - При ошибках подключения или выполнения запроса к БД.
|
||||
# @PARAM: db_config (Dict) - Конфигурация для подключения к БД.
|
||||
# @PARAM: table_name (str) - Имя таблицы.
|
||||
# @PARAM: table_schema (str) - Схема таблицы.
|
||||
# @RETURN: Dict[str, str] - Словарь с комментариями к колонкам.
|
||||
def get_postgres_comments(self, db_config: Dict, table_name: str, table_schema: str) -> Dict[str, str]:
|
||||
self.logger.info("[get_postgres_comments][Enter] Fetching comments from PostgreSQL for %s.%s.", table_schema, table_name)
|
||||
query = f"""
|
||||
@@ -84,15 +85,15 @@ class DatasetMapper:
|
||||
self.logger.error("[get_postgres_comments][Failure] %s", e, exc_info=True)
|
||||
raise
|
||||
return comments
|
||||
# </ANCHOR id="DatasetMapper.get_postgres_comments">
|
||||
|
||||
# <ANCHOR id="DatasetMapper.load_excel_mappings" type="Function">
|
||||
# @PURPOSE: Загружает меппинги 'column_name' -> 'column_comment' из XLSX файла.
|
||||
# @PRE: `file_path` должен быть валидным путем к XLSX файлу с колонками 'column_name' и 'column_comment'.
|
||||
# @POST: Возвращается словарь с меппингами.
|
||||
# @PARAM: file_path: str - Путь к XLSX файлу.
|
||||
# @RETURN: Dict[str, str] - Словарь с меппингами.
|
||||
# @THROW: Exception - При ошибках чтения файла или парсинга.
|
||||
# [/DEF:DatasetMapper.get_postgres_comments]
|
||||
|
||||
# [DEF:DatasetMapper.load_excel_mappings:Function]
|
||||
# @PURPOSE: Загружает меппинги 'column_name' -> 'column_comment' из XLSX файла.
|
||||
# @PRE: `file_path` должен быть валидным путем к XLSX файлу с колонками 'column_name' и 'column_comment'.
|
||||
# @POST: Возвращается словарь с меппингами.
|
||||
# @THROW: Exception - При ошибках чтения файла или парсинга.
|
||||
# @PARAM: file_path (str) - Путь к XLSX файлу.
|
||||
# @RETURN: Dict[str, str] - Словарь с меппингами.
|
||||
def load_excel_mappings(self, file_path: str) -> Dict[str, str]:
|
||||
self.logger.info("[load_excel_mappings][Enter] Loading mappings from %s.", file_path)
|
||||
try:
|
||||
@@ -103,21 +104,21 @@ class DatasetMapper:
|
||||
except Exception as e:
|
||||
self.logger.error("[load_excel_mappings][Failure] %s", e, exc_info=True)
|
||||
raise
|
||||
# </ANCHOR id="DatasetMapper.load_excel_mappings">
|
||||
|
||||
# <ANCHOR id="DatasetMapper.run_mapping" type="Function">
|
||||
# @PURPOSE: Основная функция для выполнения меппинга и обновления verbose_map датасета в Superset.
|
||||
# @PARAM: superset_client: SupersetClient - Клиент Superset.
|
||||
# @PARAM: dataset_id: int - ID датасета для обновления.
|
||||
# @PARAM: source: str - Источник данных ('postgres', 'excel', 'both').
|
||||
# @PARAM: postgres_config: Optional[Dict] - Конфигурация для подключения к PostgreSQL.
|
||||
# @PARAM: excel_path: Optional[str] - Путь к XLSX файлу.
|
||||
# @PARAM: table_name: Optional[str] - Имя таблицы в PostgreSQL.
|
||||
# @PARAM: table_schema: Optional[str] - Схема таблицы в PostgreSQL.
|
||||
# @RELATION: CALLS -> self.get_postgres_comments
|
||||
# @RELATION: CALLS -> self.load_excel_mappings
|
||||
# @RELATION: CALLS -> superset_client.get_dataset
|
||||
# @RELATION: CALLS -> superset_client.update_dataset
|
||||
# [/DEF:DatasetMapper.load_excel_mappings]
|
||||
|
||||
# [DEF:DatasetMapper.run_mapping:Function]
|
||||
# @PURPOSE: Основная функция для выполнения меппинга и обновления verbose_map датасета в Superset.
|
||||
# @RELATION: CALLS -> self.get_postgres_comments
|
||||
# @RELATION: CALLS -> self.load_excel_mappings
|
||||
# @RELATION: CALLS -> superset_client.get_dataset
|
||||
# @RELATION: CALLS -> superset_client.update_dataset
|
||||
# @PARAM: superset_client (SupersetClient) - Клиент Superset.
|
||||
# @PARAM: dataset_id (int) - ID датасета для обновления.
|
||||
# @PARAM: source (str) - Источник данных ('postgres', 'excel', 'both').
|
||||
# @PARAM: postgres_config (Optional[Dict]) - Конфигурация для подключения к PostgreSQL.
|
||||
# @PARAM: excel_path (Optional[str]) - Путь к XLSX файлу.
|
||||
# @PARAM: table_name (Optional[str]) - Имя таблицы в PostgreSQL.
|
||||
# @PARAM: table_schema (Optional[str]) - Схема таблицы в PostgreSQL.
|
||||
def run_mapping(self, superset_client: SupersetClient, dataset_id: int, source: str, postgres_config: Optional[Dict] = None, excel_path: Optional[str] = None, table_name: Optional[str] = None, table_schema: Optional[str] = None):
|
||||
self.logger.info("[run_mapping][Enter] Starting dataset mapping for ID %d from source '%s'.", dataset_id, source)
|
||||
mappings: Dict[str, str] = {}
|
||||
@@ -132,14 +133,14 @@ class DatasetMapper:
|
||||
if source not in ['postgres', 'excel', 'both']:
|
||||
self.logger.error("[run_mapping][Failure] Invalid source: %s.", source)
|
||||
return
|
||||
|
||||
|
||||
dataset_response = superset_client.get_dataset(dataset_id)
|
||||
dataset_data = dataset_response['result']
|
||||
|
||||
original_columns = dataset_data.get('columns', [])
|
||||
updated_columns = []
|
||||
changes_made = False
|
||||
|
||||
|
||||
for column in original_columns:
|
||||
col_name = column.get('column_name')
|
||||
|
||||
@@ -161,7 +162,7 @@ class DatasetMapper:
|
||||
}
|
||||
|
||||
new_column = {k: v for k, v in new_column.items() if v is not None}
|
||||
|
||||
|
||||
if col_name in mappings:
|
||||
mapping_value = mappings[col_name]
|
||||
if isinstance(mapping_value, str) and new_column.get('verbose_name') != mapping_value:
|
||||
@@ -169,7 +170,7 @@ class DatasetMapper:
|
||||
changes_made = True
|
||||
|
||||
updated_columns.append(new_column)
|
||||
|
||||
|
||||
updated_metrics = []
|
||||
for metric in dataset_data.get("metrics", []):
|
||||
new_metric = {
|
||||
@@ -186,7 +187,7 @@ class DatasetMapper:
|
||||
"uuid": metric.get("uuid"),
|
||||
}
|
||||
updated_metrics.append({k: v for k, v in new_metric.items() if v is not None})
|
||||
|
||||
|
||||
if changes_made:
|
||||
payload_for_update = {
|
||||
"database_id": dataset_data.get("database", {}).get("id"),
|
||||
@@ -213,18 +214,16 @@ class DatasetMapper:
|
||||
}
|
||||
|
||||
payload_for_update = {k: v for k, v in payload_for_update.items() if v is not None}
|
||||
|
||||
|
||||
superset_client.update_dataset(dataset_id, payload_for_update)
|
||||
self.logger.info("[run_mapping][Success] Dataset %d columns' verbose_name updated.", dataset_id)
|
||||
else:
|
||||
self.logger.info("[run_mapping][State] No changes in columns' verbose_name, skipping update.")
|
||||
|
||||
|
||||
except (AssertionError, FileNotFoundError, Exception) as e:
|
||||
self.logger.error("[run_mapping][Failure] %s", e, exc_info=True)
|
||||
return
|
||||
# </ANCHOR id="DatasetMapper.run_mapping">
|
||||
# </ANCHOR id="DatasetMapper">
|
||||
|
||||
# --- Конец кода модуля ---
|
||||
|
||||
# </GRACE_MODULE id="dataset_mapper">
|
||||
# [/DEF:DatasetMapper.run_mapping]
|
||||
# [/DEF:DatasetMapper]
|
||||
|
||||
# [/DEF:superset_tool.utils.dataset_mapper]
|
||||
|
||||
Reference in New Issue
Block a user