90 lines
5.0 KiB
Python
90 lines
5.0 KiB
Python
# pylint: disable=no-self-argument,too-few-public-methods
|
||
"""
|
||
[MODULE] Сущности данных конфигурации
|
||
@desc: Определяет структуры данных, используемые для конфигурации и трансформации в инструменте Superset.
|
||
"""
|
||
|
||
# [IMPORTS] Pydantic и Typing
|
||
from typing import Optional, Dict, Any
|
||
from pydantic import BaseModel, validator, Field, HttpUrl, VERSION
|
||
|
||
# [IMPORTS] Локальные модули
|
||
from .utils.logger import SupersetLogger
|
||
|
||
class SupersetConfig(BaseModel):
|
||
"""
|
||
[CONFIG] Конфигурация подключения к Superset API.
|
||
"""
|
||
base_url: str = Field(..., description="Базовый URL Superset API, включая версию /api/v1.", pattern=r'.*/api/v1.*')
|
||
auth: Dict[str, str] = Field(..., description="Словарь с данными для аутентификации (provider, username, password, refresh).")
|
||
verify_ssl: bool = Field(True, description="Флаг для проверки SSL-сертификатов.")
|
||
timeout: int = Field(30, description="Таймаут в секундах для HTTP-запросов.")
|
||
logger: Optional[SupersetLogger] = Field(None, description="Экземпляр логгера для логирования внутри клиента.")
|
||
|
||
# [ENTITY: Function('validate_auth')]
|
||
# CONTRACT:
|
||
# PURPOSE: Валидация словаря `auth`.
|
||
# PRECONDITIONS: `v` должен быть словарем.
|
||
# POSTCONDITIONS: Возвращает `v` если все обязательные поля присутствуют.
|
||
@validator('auth')
|
||
def validate_auth(cls, v: Dict[str, str], values: dict) -> Dict[str, str]:
|
||
logger = values.get('logger') or SupersetLogger(name="SupersetConfig")
|
||
logger.debug("[DEBUG][SupersetConfig.validate_auth][ENTER] Validating auth.")
|
||
required = {'provider', 'username', 'password', 'refresh'}
|
||
if not required.issubset(v.keys()):
|
||
logger.error("[ERROR][SupersetConfig.validate_auth][FAILURE] Missing required auth fields.")
|
||
raise ValueError(f"Словарь 'auth' должен содержать поля: {required}. Отсутствующие: {required - v.keys()}")
|
||
logger.debug("[DEBUG][SupersetConfig.validate_auth][SUCCESS] Auth validated.")
|
||
return v
|
||
# END_FUNCTION_validate_auth
|
||
|
||
# [ENTITY: Function('check_base_url_format')]
|
||
# CONTRACT:
|
||
# PURPOSE: Валидация формата `base_url`.
|
||
# PRECONDITIONS: `v` должна быть строкой.
|
||
# POSTCONDITIONS: Возвращает `v` если это валидный URL.
|
||
@validator('base_url')
|
||
def check_base_url_format(cls, v: str, values: dict) -> str:
|
||
logger = values.get('logger') or SupersetLogger(name="SupersetConfig")
|
||
logger.debug("[DEBUG][SupersetConfig.check_base_url_format][ENTER] Validating base_url.")
|
||
try:
|
||
if VERSION.startswith('1'):
|
||
HttpUrl(v)
|
||
except (ValueError, TypeError) as exc:
|
||
logger.error("[ERROR][SupersetConfig.check_base_url_format][FAILURE] Invalid base_url format.")
|
||
raise ValueError(f"Invalid URL format: {v}") from exc
|
||
logger.debug("[DEBUG][SupersetConfig.check_base_url_format][SUCCESS] base_url validated.")
|
||
return v
|
||
# END_FUNCTION_check_base_url_format
|
||
|
||
class Config:
|
||
"""Pydantic config"""
|
||
arbitrary_types_allowed = True
|
||
|
||
class DatabaseConfig(BaseModel):
|
||
"""
|
||
[CONFIG] Параметры трансформации баз данных при миграции дашбордов.
|
||
"""
|
||
database_config: Dict[str, Dict[str, Any]] = Field(..., description="Словарь, содержащий 'old' и 'new' конфигурации базы данных.")
|
||
logger: Optional[SupersetLogger] = Field(None, description="Экземпляр логгера для логирования.")
|
||
|
||
# [ENTITY: Function('validate_config')]
|
||
# CONTRACT:
|
||
# PURPOSE: Валидация словаря `database_config`.
|
||
# PRECONDITIONS: `v` должен быть словарем.
|
||
# POSTCONDITIONS: Возвращает `v` если содержит ключи 'old' и 'new'.
|
||
@validator('database_config')
|
||
def validate_config(cls, v: Dict[str, Dict[str, Any]], values: dict) -> Dict[str, Dict[str, Any]]:
|
||
logger = values.get('logger') or SupersetLogger(name="DatabaseConfig")
|
||
logger.debug("[DEBUG][DatabaseConfig.validate_config][ENTER] Validating database_config.")
|
||
if not {'old', 'new'}.issubset(v.keys()):
|
||
logger.error("[ERROR][DatabaseConfig.validate_config][FAILURE] Missing 'old' or 'new' keys in database_config.")
|
||
raise ValueError("'database_config' должен содержать ключи 'old' и 'new'.")
|
||
logger.debug("[DEBUG][DatabaseConfig.validate_config][SUCCESS] database_config validated.")
|
||
return v
|
||
# END_FUNCTION_validate_config
|
||
|
||
class Config:
|
||
"""Pydantic config"""
|
||
arbitrary_types_allowed = True
|