service/src/utils/settings.py - master - aastray/StudentPart

service/src/utils/settings.py
99 строк3 KB
aastray
Реализован функционал загрузки файлов и просмотра документов, удалены базовые эндпоинты, обновлена документация
10 янв 2026, 15:38
10 янв 2026, 15:384d47c55
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
import os
from pathlib import Path
from typing import List, Dict, Optional, Any
import json

env = os.getenv("ENVIRONMENT")

# Определяем путь к .env файлу (в корне проекта)
# service/src/utils/settings.py -> service/src/utils -> service/src -> service -> корень проекта
_current_file = Path(__file__).resolve()  # Получаем абсолютный путь
_project_root = _current_file.parent.parent.parent.parent  # service/src/utils -> service/src -> service -> корень
_env_file_path = _project_root / ".env"

# Для Docker: .env может быть в /app
if not _env_file_path.exists():
    docker_env = Path("/app/.env")
    if docker_env.exists():
        _env_file_path = docker_env

# Если все еще не найден, используем относительный путь (pydantic-settings будет искать сам)
if not _env_file_path.exists():
    _env_file_path = Path(".env")  # Будет искаться относительно текущей рабочей директории


class Settings(BaseSettings):

    DB_USERNAME: str
    DB_PASSWORD: str
    DB_NAME: str
    DB_HOST: str
    DB_PORT: int = 5432
    DT_INFO: Dict[str, str]
    DT_ENDPOINTS: List[Dict[str, str]]
    DT_FIRST_FORM: List[Dict[str, str]]

    ai_api_key: Optional[str] = None
    ai_model: Optional[str] = None

    @field_validator('DT_INFO', mode='before')
    @classmethod
    def parse_dt_info(cls, value):
        if isinstance(value, str):
            try:
                return json.loads(value)
            except json.JSONDecodeError:
                if 'dt_info' in cls.__annotations__ and cls.__annotations__['dt_info'] == Dict[str, str]:
                    return {}
                else:
                    return []
        return value
    
    model_config = SettingsConfigDict(
        env_file=str(_env_file_path) if _env_file_path.exists() else '.env',
        env_file_encoding='utf-8',
        env_nested_delimiter='__',
        extra="allow",
        case_sensitive=True
    )

    #Для FastAPI
    def get_database_url_async(self):
        db_host = self._get_db_host()
        return (f"postgresql+asyncpg://"
                f"{self.DB_USERNAME}:"
                f"{self.DB_PASSWORD}@"
                f"{db_host}:"
                f"{self.DB_PORT}/"
                f"{self.DB_NAME}")
    
    def _get_db_host(self) -> str:
        """
        Автоматически определяет правильный хост БД.
        Если указан postgres_db (Docker имя), но мы не в Docker, использует localhost.
        """
        host = self.DB_HOST
        
        # Если хост - имя Docker сервиса, проверяем, в Docker ли мы
        if host == "postgres_db" or host == "postgres":
            # Проверяем, находимся ли мы в Docker (есть /app)
            if not Path("/app").exists():
                # Не в Docker, используем localhost
                return "localhost"
        
        return host
    
    #Для Alembic
    def get_database_url_sync(self):
        db_host = self._get_db_host()
        return (f"postgresql://"
            f"{self.DB_USERNAME}:"
            f"{self.DB_PASSWORD}@"
            f"{db_host}:"
            f"{self.DB_PORT}/"
            f"{self.DB_NAME}")


settings = Settings()