147 lines
8.1 KiB
Python
147 lines
8.1 KiB
Python
# [MODULE] Сущности данных конфигурации
|
||
# @desc: Определяет структуры данных, используемые для конфигурации и трансформации в инструменте Superset.
|
||
# @contracts:
|
||
# - Все модели наследуются от `pydantic.BaseModel` для автоматической валидации.
|
||
# - Валидация URL-адресов и параметров аутентификации.
|
||
# - Валидация структуры конфигурации БД для миграций.
|
||
# @coherence:
|
||
# - Все модели согласованы со схемой API Superset v1.
|
||
# - Совместимы с клиентскими методами `SupersetClient` и утилитами.
|
||
|
||
# [IMPORTS] Pydantic и Typing
|
||
from typing import Optional, Dict, Any, Union
|
||
from pydantic import BaseModel, validator, Field, HttpUrl
|
||
# [COHERENCE_CHECK_PASSED] Все необходимые импорты для Pydantic моделей.
|
||
|
||
# [IMPORTS] Локальные модули
|
||
from .utils.logger import SupersetLogger
|
||
|
||
class SupersetConfig(BaseModel):
|
||
"""[CONFIG] Конфигурация подключения к Superset API.
|
||
@semantic: Инкапсулирует основные параметры, необходимые для инициализации `SupersetClient`.
|
||
@invariant:
|
||
- `base_url` должен быть валидным HTTP(S) URL и содержать `/api/v1`.
|
||
- `auth` должен содержать обязательные поля для аутентификации по логину/паролю.
|
||
- `timeout` должен быть положительным числом.
|
||
"""
|
||
base_url: str = Field(..., description="Базовый URL Superset API, включая версию /api/v1.", regex=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="Экземпляр логгера для логирования внутри клиента.")
|
||
|
||
# [VALIDATOR] Проверка параметров аутентификации
|
||
@validator('auth')
|
||
def validate_auth(cls, v: Dict[str, str]) -> Dict[str, str]:
|
||
"""[CONTRACT_VALIDATOR] Валидация словаря `auth`.
|
||
@pre:
|
||
- `v` должен быть словарем.
|
||
@post:
|
||
- Возвращает `v` если все обязательные поля присутствуют.
|
||
@raise:
|
||
- `ValueError`: Если отсутствуют обязательные поля ('provider', 'username', 'password', 'refresh').
|
||
"""
|
||
required = {'provider', 'username', 'password', 'refresh'}
|
||
if not required.issubset(v.keys()):
|
||
raise ValueError(
|
||
f"[CONTRACT_VIOLATION] Словарь 'auth' должен содержать поля: {required}. "
|
||
f"Отсутствующие: {required - v.keys()}"
|
||
)
|
||
# [COHERENCE_CHECK_PASSED] Auth-конфигурация валидна.
|
||
return v
|
||
|
||
# [VALIDATOR] Проверка base_url
|
||
@validator('base_url')
|
||
def check_base_url_format(cls, v: str) -> str:
|
||
"""[CONTRACT_VALIDATOR] Валидация формата `base_url`.
|
||
@pre:
|
||
- `v` должна быть строкой.
|
||
@post:
|
||
- Возвращает `v` если это валидный URL.
|
||
@raise:
|
||
- `ValueError`: Если URL невалиден.
|
||
"""
|
||
try:
|
||
# Для Pydantic v2:
|
||
from pydantic import HttpUrl
|
||
HttpUrl(v, scheme="https") # Явное указание схемы
|
||
except ValueError:
|
||
# Для совместимости с Pydantic v1:
|
||
HttpUrl(v)
|
||
return v
|
||
|
||
class Config:
|
||
arbitrary_types_allowed = True # Разрешаем Pydantic обрабатывать произвольные типы (например, SupersetLogger)
|
||
json_schema_extra = {
|
||
"example": {
|
||
"base_url": "https://host/api/v1/",
|
||
"auth": {
|
||
"provider": "db",
|
||
"username": "user",
|
||
"password": "pass",
|
||
"refresh": True
|
||
},
|
||
"verify_ssl": True,
|
||
"timeout": 60
|
||
}
|
||
}
|
||
|
||
|
||
# [SEMANTIC-TYPE] Конфигурация БД для миграций
|
||
class DatabaseConfig(BaseModel):
|
||
"""[CONFIG] Параметры трансформации баз данных при миграции дашбордов.
|
||
@semantic: Содержит `old` и `new` состояния конфигурации базы данных,
|
||
используемые для поиска и замены в YAML-файлах экспортированных дашбордов.
|
||
@invariant:
|
||
- `database_config` должен быть словарем с ключами 'old' и 'new'.
|
||
- Каждое из 'old' и 'new' должно быть словарем, содержащим метаданные БД Superset.
|
||
"""
|
||
database_config: Dict[str, Dict[str, Any]] = Field(..., description="Словарь, содержащий 'old' и 'new' конфигурации базы данных.")
|
||
logger: Optional[SupersetLogger] = Field(None, description="Экземпляр логгера для логирования.")
|
||
|
||
@validator('database_config')
|
||
def validate_config(cls, v: Dict[str, Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
|
||
"""[CONTRACT_VALIDATOR] Валидация словаря `database_config`.
|
||
@pre:
|
||
- `v` должен быть словарем.
|
||
@post:
|
||
- Возвращает `v` если содержит ключи 'old' и 'new'.
|
||
@raise:
|
||
- `ValueError`: Если отсутствуют ключи 'old' или 'new'.
|
||
"""
|
||
if not {'old', 'new'}.issubset(v.keys()):
|
||
raise ValueError(
|
||
"[CONTRACT_VIOLATION] 'database_config' должен содержать ключи 'old' и 'new'."
|
||
)
|
||
# Дополнительно можно добавить проверку структуры `old` и `new` на наличие `uuid`, `database_name` и т.д.
|
||
# Для простоты пока ограничимся наличием ключей 'old' и 'new'.
|
||
# [COHERENCE_CHECK_PASSED] Конфигурация базы данных для миграции валидна.
|
||
return v
|
||
|
||
class Config:
|
||
arbitrary_types_allowed = True
|
||
json_schema_extra = {
|
||
"example": {
|
||
"database_config": {
|
||
"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"
|
||
}
|
||
}
|
||
}
|
||
} |