356 lines
14 KiB
Python
356 lines
14 KiB
Python
import os
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Dict, List, Literal, Optional
|
||
|
||
from services.redis import redis
|
||
from utils.logger import root_logger as logger
|
||
|
||
|
||
@dataclass
|
||
class EnvVariable:
|
||
"""Представление переменной окружения"""
|
||
|
||
key: str
|
||
value: str = ""
|
||
description: str = ""
|
||
type: Literal["string", "integer", "boolean", "json"] = "string" # string, integer, boolean, json
|
||
is_secret: bool = False
|
||
|
||
|
||
@dataclass
|
||
class EnvSection:
|
||
"""Группа переменных окружения"""
|
||
|
||
name: str
|
||
description: str
|
||
variables: List[EnvVariable]
|
||
|
||
|
||
class EnvManager:
|
||
"""
|
||
Менеджер переменных окружения с поддержкой Redis кеширования
|
||
"""
|
||
|
||
# Определение секций с их описаниями
|
||
SECTIONS = {
|
||
"database": "Настройки базы данных",
|
||
"auth": "Настройки аутентификации",
|
||
"redis": "Настройки Redis",
|
||
"search": "Настройки поиска",
|
||
"integrations": "Внешние интеграции",
|
||
"security": "Настройки безопасности",
|
||
"logging": "Настройки логирования",
|
||
"features": "Флаги функций",
|
||
"other": "Прочие настройки",
|
||
}
|
||
|
||
# Маппинг переменных на секции
|
||
VARIABLE_SECTIONS = {
|
||
# Database
|
||
"DB_URL": "database",
|
||
"DATABASE_URL": "database",
|
||
"POSTGRES_USER": "database",
|
||
"POSTGRES_PASSWORD": "database",
|
||
"POSTGRES_DB": "database",
|
||
"POSTGRES_HOST": "database",
|
||
"POSTGRES_PORT": "database",
|
||
# Auth
|
||
"JWT_SECRET": "auth",
|
||
"JWT_ALGORITHM": "auth",
|
||
"JWT_EXPIRATION": "auth",
|
||
"SECRET_KEY": "auth",
|
||
"AUTH_SECRET": "auth",
|
||
"OAUTH_GOOGLE_CLIENT_ID": "auth",
|
||
"OAUTH_GOOGLE_CLIENT_SECRET": "auth",
|
||
"OAUTH_GITHUB_CLIENT_ID": "auth",
|
||
"OAUTH_GITHUB_CLIENT_SECRET": "auth",
|
||
# Redis
|
||
"REDIS_URL": "redis",
|
||
"REDIS_HOST": "redis",
|
||
"REDIS_PORT": "redis",
|
||
"REDIS_PASSWORD": "redis",
|
||
"REDIS_DB": "redis",
|
||
# Search
|
||
"SEARCH_API_KEY": "search",
|
||
"ELASTICSEARCH_URL": "search",
|
||
"SEARCH_INDEX": "search",
|
||
# Integrations
|
||
"GOOGLE_ANALYTICS_ID": "integrations",
|
||
"SENTRY_DSN": "integrations",
|
||
"SMTP_HOST": "integrations",
|
||
"SMTP_PORT": "integrations",
|
||
"SMTP_USER": "integrations",
|
||
"SMTP_PASSWORD": "integrations",
|
||
"EMAIL_FROM": "integrations",
|
||
# Security
|
||
"CORS_ORIGINS": "security",
|
||
"ALLOWED_HOSTS": "security",
|
||
"SECURE_SSL_REDIRECT": "security",
|
||
"SESSION_COOKIE_SECURE": "security",
|
||
"CSRF_COOKIE_SECURE": "security",
|
||
# Logging
|
||
"LOG_LEVEL": "logging",
|
||
"LOG_FORMAT": "logging",
|
||
"LOG_FILE": "logging",
|
||
"DEBUG": "logging",
|
||
# Features
|
||
"FEATURE_REGISTRATION": "features",
|
||
"FEATURE_COMMENTS": "features",
|
||
"FEATURE_ANALYTICS": "features",
|
||
"FEATURE_SEARCH": "features",
|
||
}
|
||
|
||
# Секретные переменные (не показываем их значения в UI)
|
||
SECRET_VARIABLES = {
|
||
"JWT_SECRET",
|
||
"SECRET_KEY",
|
||
"AUTH_SECRET",
|
||
"OAUTH_GOOGLE_CLIENT_SECRET",
|
||
"OAUTH_GITHUB_CLIENT_SECRET",
|
||
"POSTGRES_PASSWORD",
|
||
"REDIS_PASSWORD",
|
||
"SEARCH_API_KEY",
|
||
"SENTRY_DSN",
|
||
"SMTP_PASSWORD",
|
||
}
|
||
|
||
def __init__(self) -> None:
|
||
self.redis_prefix = "env_vars:"
|
||
|
||
def _get_variable_type(self, key: str, value: str) -> Literal["string", "integer", "boolean", "json"]:
|
||
"""Определяет тип переменной на основе ключа и значения"""
|
||
|
||
# Boolean переменные
|
||
if value.lower() in ("true", "false", "1", "0", "yes", "no"):
|
||
return "boolean"
|
||
|
||
# Integer переменные
|
||
if key.endswith(("_PORT", "_TIMEOUT", "_LIMIT", "_SIZE")) or value.isdigit():
|
||
return "integer"
|
||
|
||
# JSON переменные
|
||
if value.startswith(("{", "[")) and value.endswith(("}", "]")):
|
||
return "json"
|
||
|
||
return "string"
|
||
|
||
def _get_variable_description(self, key: str) -> str:
|
||
"""Генерирует описание для переменной на основе её ключа"""
|
||
|
||
descriptions = {
|
||
"DB_URL": "URL подключения к базе данных",
|
||
"REDIS_URL": "URL подключения к Redis",
|
||
"JWT_SECRET": "Секретный ключ для подписи JWT токенов",
|
||
"CORS_ORIGINS": "Разрешенные CORS домены",
|
||
"DEBUG": "Режим отладки (true/false)",
|
||
"LOG_LEVEL": "Уровень логирования (DEBUG, INFO, WARNING, ERROR)",
|
||
"SENTRY_DSN": "DSN для интеграции с Sentry",
|
||
"GOOGLE_ANALYTICS_ID": "ID для Google Analytics",
|
||
"OAUTH_GOOGLE_CLIENT_ID": "Client ID для OAuth Google",
|
||
"OAUTH_GOOGLE_CLIENT_SECRET": "Client Secret для OAuth Google",
|
||
"OAUTH_GITHUB_CLIENT_ID": "Client ID для OAuth GitHub",
|
||
"OAUTH_GITHUB_CLIENT_SECRET": "Client Secret для OAuth GitHub",
|
||
"SMTP_HOST": "SMTP сервер для отправки email",
|
||
"SMTP_PORT": "Порт SMTP сервера",
|
||
"SMTP_USER": "Пользователь SMTP",
|
||
"SMTP_PASSWORD": "Пароль SMTP",
|
||
"EMAIL_FROM": "Email отправителя по умолчанию",
|
||
}
|
||
|
||
return descriptions.get(key, f"Переменная окружения {key}")
|
||
|
||
async def get_variables_from_redis(self) -> Dict[str, str]:
|
||
"""Получает переменные из Redis"""
|
||
|
||
try:
|
||
# Get all keys matching our prefix
|
||
pattern = f"{self.redis_prefix}*"
|
||
keys = await redis.execute("KEYS", pattern)
|
||
|
||
if not keys:
|
||
return {}
|
||
|
||
redis_vars: Dict[str, str] = {}
|
||
for key in keys:
|
||
var_key = key.replace(self.redis_prefix, "")
|
||
value = await redis.get(key)
|
||
if value:
|
||
if isinstance(value, bytes):
|
||
redis_vars[var_key] = value.decode("utf-8")
|
||
else:
|
||
redis_vars[var_key] = str(value)
|
||
|
||
return redis_vars
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении переменных из Redis: {e}")
|
||
return {}
|
||
|
||
async def set_variables_to_redis(self, variables: Dict[str, str]) -> bool:
|
||
"""Сохраняет переменные в Redis"""
|
||
|
||
try:
|
||
for key, value in variables.items():
|
||
redis_key = f"{self.redis_prefix}{key}"
|
||
await redis.set(redis_key, value)
|
||
|
||
logger.info(f"Сохранено {len(variables)} переменных в Redis")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при сохранении переменных в Redis: {e}")
|
||
return False
|
||
|
||
def get_variables_from_env(self) -> Dict[str, str]:
|
||
"""Получает переменные из системного окружения"""
|
||
|
||
env_vars = {}
|
||
|
||
# Получаем все переменные известные системе
|
||
for key in self.VARIABLE_SECTIONS.keys():
|
||
value = os.getenv(key)
|
||
if value is not None:
|
||
env_vars[key] = value
|
||
|
||
# Также ищем переменные по паттернам
|
||
for env_key, env_value in os.environ.items():
|
||
# Переменные проекта обычно начинаются с определенных префиксов
|
||
if any(env_key.startswith(prefix) for prefix in ["APP_", "SITE_", "FEATURE_", "OAUTH_"]):
|
||
env_vars[env_key] = env_value
|
||
|
||
return env_vars
|
||
|
||
async def get_all_variables(self) -> List[EnvSection]:
|
||
"""Получает все переменные окружения, сгруппированные по секциям"""
|
||
|
||
# Получаем переменные из разных источников
|
||
env_vars = self.get_variables_from_env()
|
||
redis_vars = await self.get_variables_from_redis()
|
||
|
||
# Объединяем переменные (приоритет у Redis)
|
||
all_vars = {**env_vars, **redis_vars}
|
||
|
||
# Группируем по секциям
|
||
sections_dict: Dict[str, List[EnvVariable]] = {section: [] for section in self.SECTIONS}
|
||
other_variables: List[EnvVariable] = [] # Для переменных, которые не попали ни в одну секцию
|
||
|
||
for key, value in all_vars.items():
|
||
section_name = self.VARIABLE_SECTIONS.get(key, "other")
|
||
is_secret = key in self.SECRET_VARIABLES
|
||
|
||
var = EnvVariable(
|
||
key=key,
|
||
value=value if not is_secret else "***", # Скрываем секретные значения
|
||
description=self._get_variable_description(key),
|
||
type=self._get_variable_type(key, value),
|
||
is_secret=is_secret,
|
||
)
|
||
|
||
if section_name in sections_dict:
|
||
sections_dict[section_name].append(var)
|
||
else:
|
||
other_variables.append(var)
|
||
|
||
# Добавляем переменные без секции в раздел "other"
|
||
if other_variables:
|
||
sections_dict["other"].extend(other_variables)
|
||
|
||
# Создаем объекты секций
|
||
sections = []
|
||
for section_key, variables in sections_dict.items():
|
||
if variables: # Добавляем только секции с переменными
|
||
sections.append(
|
||
EnvSection(
|
||
name=section_key,
|
||
description=self.SECTIONS[section_key],
|
||
variables=sorted(variables, key=lambda x: x.key),
|
||
)
|
||
)
|
||
|
||
return sorted(sections, key=lambda x: x.name)
|
||
|
||
async def update_variables(self, variables: List[EnvVariable]) -> bool:
|
||
"""Обновляет переменные окружения"""
|
||
|
||
try:
|
||
# Подготавливаем данные для сохранения
|
||
vars_to_save = {}
|
||
|
||
for var in variables:
|
||
# Валидация
|
||
if not var.key or not isinstance(var.key, str):
|
||
logger.error(f"Неверный ключ переменной: {var.key}")
|
||
continue
|
||
|
||
# Проверяем формат ключа (только буквы, цифры и подчеркивания)
|
||
if not re.match(r"^[A-Z_][A-Z0-9_]*$", var.key):
|
||
logger.error(f"Неверный формат ключа: {var.key}")
|
||
continue
|
||
|
||
vars_to_save[var.key] = var.value
|
||
|
||
if not vars_to_save:
|
||
logger.warning("Нет переменных для сохранения")
|
||
return False
|
||
|
||
# Сохраняем в Redis
|
||
success = await self.set_variables_to_redis(vars_to_save)
|
||
|
||
if success:
|
||
logger.info(f"Обновлено {len(vars_to_save)} переменных окружения")
|
||
|
||
return success
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при обновлении переменных: {e}")
|
||
return False
|
||
|
||
async def delete_variable(self, key: str) -> bool:
|
||
"""Удаляет переменную окружения"""
|
||
|
||
try:
|
||
redis_key = f"{self.redis_prefix}{key}"
|
||
result = await redis.delete(redis_key)
|
||
|
||
if result > 0:
|
||
logger.info(f"Переменная {key} удалена")
|
||
return True
|
||
logger.warning(f"Переменная {key} не найдена")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при удалении переменной {key}: {e}")
|
||
return False
|
||
|
||
async def get_variable(self, key: str) -> Optional[str]:
|
||
"""Получает значение конкретной переменной"""
|
||
|
||
# Сначала проверяем Redis
|
||
try:
|
||
redis_key = f"{self.redis_prefix}{key}"
|
||
value = await redis.get(redis_key)
|
||
if value:
|
||
return value.decode("utf-8") if isinstance(value, bytes) else str(value)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении переменной {key} из Redis: {e}")
|
||
|
||
# Fallback на системное окружение
|
||
return os.getenv(key)
|
||
|
||
async def set_variable(self, key: str, value: str) -> bool:
|
||
"""Устанавливает значение переменной"""
|
||
|
||
try:
|
||
redis_key = f"{self.redis_prefix}{key}"
|
||
await redis.set(redis_key, value)
|
||
logger.info(f"Переменная {key} установлена")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при установке переменной {key}: {e}")
|
||
return False
|
||
|
||
|
||
env_manager = EnvManager()
|