# [DEF:superset_tool.models:Module] # # @SEMANTICS: pydantic, model, config, validation, data-structure # @PURPOSE: Определяет Pydantic-модели для конфигурации инструмента, обеспечивая валидацию данных. # @LAYER: Infra # @RELATION: DEPENDS_ON -> pydantic # @RELATION: DEPENDS_ON -> superset_tool.utils.logger # @PUBLIC_API: SupersetConfig, DatabaseConfig # [SECTION: IMPORTS] import re from typing import Optional, Dict, Any from pydantic import BaseModel, validator, Field from .utils.logger import SupersetLogger # [/SECTION] # [DEF:SupersetConfig:Class] # @PURPOSE: Модель конфигурации для подключения к одному экземпляру Superset API. # @RELATION: INHERITS_FROM -> pydantic.BaseModel class SupersetConfig(BaseModel): env: str = Field(..., description="Название окружения (например, dev, prod).") base_url: str = Field(..., description="Базовый URL Superset API, включая /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="Экземпляр логгера для логирования.") # [DEF:SupersetConfig.validate_auth:Function] # @PURPOSE: Проверяет, что словарь `auth` содержит все необходимые для аутентификации поля. # @PRE: `v` должен быть словарем. # @POST: Возвращает `v`, если все обязательные поля (`provider`, `username`, `password`, `refresh`) присутствуют. # @THROW: ValueError - Если отсутствуют обязательные поля. # @PARAM: v (Dict[str, str]) - Значение поля auth. @validator('auth') def validate_auth(cls, v: Dict[str, str]) -> Dict[str, str]: required = {'provider', 'username', 'password', 'refresh'} if not required.issubset(v.keys()): raise ValueError(f"Словарь 'auth' должен содержать поля: {required}. Отсутствующие: {required - v.keys()}") return v # [/DEF:SupersetConfig.validate_auth] # [DEF:SupersetConfig.normalize_base_url:Function] # @PURPOSE: Нормализует `base_url`, добавляя `/api/v1`, если он отсутствует. # @PRE: `v` должна быть строкой. # @POST: Возвращает нормализованный `v`. # @THROW: ValueError - Если формат URL невалиден. # @PARAM: v (str) - Значение поля base_url. @validator('base_url') def normalize_base_url(cls, v: str) -> str: v = v.strip() if not v.startswith(('http://', 'https://')): raise ValueError(f"Invalid URL scheme: {v}. Must start with http:// or https://") if '/api/v1' not in v: v = f"{v.rstrip('/')}/api/v1" return v # [/DEF:SupersetConfig.normalize_base_url] class Config: arbitrary_types_allowed = True # [/DEF:SupersetConfig] # [DEF:DatabaseConfig:Class] # @PURPOSE: Модель для параметров трансформации баз данных при миграции дашбордов. # @RELATION: INHERITS_FROM -> pydantic.BaseModel class DatabaseConfig(BaseModel): database_config: Dict[str, Dict[str, Any]] = Field(..., description="Словарь, содержащий 'old' и 'new' конфигурации базы данных.") logger: Optional[SupersetLogger] = Field(None, description="Экземпляр логгера для логирования.") # [DEF:DatabaseConfig.validate_config:Function] # @PURPOSE: Проверяет, что словарь `database_config` содержит ключи 'old' и 'new'. # @PRE: `v` должен быть словарем. # @POST: Возвращает `v`, если ключи 'old' и 'new' присутствуют. # @THROW: ValueError - Если отсутствуют обязательные ключи. # @PARAM: v (Dict[str, Dict[str, Any]]) - Значение поля database_config. @validator('database_config') def validate_config(cls, v: Dict[str, Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: if not {'old', 'new'}.issubset(v.keys()): raise ValueError("'database_config' должен содержать ключи 'old' и 'new'.") return v # [/DEF:DatabaseConfig.validate_config] class Config: arbitrary_types_allowed = True # [/DEF:DatabaseConfig] # [/DEF:superset_tool.models]