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()