feat: migrate config to pydantic-settings (#005)

.github/workflows/ci.yml

@@ -21,7 +21,7 @@ permissions:
 
 # Переменные окружения
 env:
-  PYTHON_VERSION: '3.14'
+  PYTHON_VERSION: '3.13'
 
 jobs:
   # Job 1: Проверка кода

.github/workflows/quick-check.yml

@@ -26,7 +26,7 @@ permissions:
   pull-requests: write
 
 env:
-  PYTHON_VERSION: '3.14'
+  PYTHON_VERSION: '3.13'
 
 jobs:
   quick-check:

bot/config.py

@@ -1,144 +1,175 @@
-"""Bot configuration"""
+"""Bot configuration using pydantic-settings.
+
+Validates all settings at startup with clear error messages.
+Replaces manual os.getenv() calls with typed, validated fields.
+"""
 
-import os
-from dataclasses import dataclass
 from pathlib import Path
+from urllib.parse import quote_plus
 
-from dotenv import load_dotenv
+from pydantic import Field, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
 
-load_dotenv()
 
+class BotConfig(BaseSettings):
+    """Telegram bot credentials."""
 
-@dataclass
-class BotConfig:
-    """Bot configuration class"""
+    model_config = SettingsConfigDict(env_prefix="")
 
-    token: str
-    admin_ids: list[int]
+    token: str = Field(alias="BOT_TOKEN")
+    admin_ids: list[int] = Field(default_factory=list, alias="ADMIN_IDS")
 
+    @field_validator("admin_ids", mode="before")
+    @classmethod
+    def parse_admin_ids(cls, v: object) -> object:
+        if isinstance(v, str):
+            return [int(x.strip()) for x in v.split(",") if x.strip()]
+        if isinstance(v, int):
+            return [v]
+        return v
 
-@dataclass
-class Config:
-    """Main configuration class"""
 
-    bot: BotConfig
-    locales_path: Path
-    default_locale: str
-    app_url: str
-    site_url: str
-    redirect_port: int
-    db_dsn: str
-    redis_url: str
-    database_url: str
-    event_bus_mode: str
-    event_bus_consumer_name: str
-    event_store_enabled: bool
-    cache_warming_enabled: bool
-    qr_size_small: int
-    qr_size_medium: int
-    qr_size_large: int
-    # Webhook configuration
-    bot_mode: str
-    webhook_host: str
-    webhook_port: int
-    webhook_path: str
-    webhook_secret: str
-    webhook_url: str
+class DatabaseConfig(BaseSettings):
+    """PostgreSQL connection settings."""
 
+    model_config = SettingsConfigDict(env_prefix="DB_")
 
-def load_config() -> Config:
-    """Load configuration from environment"""
-    bot_token = os.getenv("BOT_TOKEN")
-    if not bot_token:
-        raise ValueError("BOT_TOKEN is not set in environment variables")
-
-    admin_ids_str = os.getenv("ADMIN_IDS", "")
-    admin_ids = [int(x.strip()) for x in admin_ids_str.split(",") if x.strip()]
-
-    db_host = os.getenv("DB_HOST", "localhost")
-    db_port = os.getenv("DB_PORT", "5432")
-    db_name = os.getenv("DB_NAME", "qr_bot")
-    db_user = os.getenv("DB_USER", "qr_bot_user")
-    db_password = os.getenv("DB_PASSWORD", "")
-
-    db_dsn = f"postgresql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}"
-
-    # Redis configuration
-    redis_host = os.getenv("REDIS_HOST", "localhost")
-    redis_port = os.getenv("REDIS_PORT", "6379")
-    redis_db = os.getenv("REDIS_DB", "0")
-    redis_password = os.getenv("REDIS_PASSWORD", "")
-    if redis_password:
-        redis_url = f"redis://:{redis_password}@{redis_host}:{redis_port}/{redis_db}"
-    else:
-        redis_url = f"redis://{redis_host}:{redis_port}/{redis_db}"
-
-    # Event Bus configuration
-    event_bus_mode = os.getenv("EVENT_BUS_MODE", "redis")  # 'memory' or 'redis'
-    event_bus_consumer_name = os.getenv(
-        "EVENT_BUS_CONSUMER_NAME", ""
-    )  # Auto-generated if empty
-
-    # Event Store configuration (PostgreSQL persistence)
-    event_store_enabled = os.getenv("EVENT_STORE_ENABLED", "true").lower() in (
-        "true",
-        "1",
-        "yes",
-    )
+    host: str = "localhost"
+    port: int = 5432
+    name: str = "qr_bot"
+    user: str = "qr_bot_user"
+    password: str = ""
 
-    # Cache warming configuration
-    cache_warming_enabled = os.getenv("CACHE_WARMING_ENABLED", "true").lower() in (
-        "true",
-        "1",
-        "yes",
-    )
+    @property
+    def dsn(self) -> str:
+        """Build PostgreSQL DSN with proper URL encoding."""
+        password = quote_plus(self.password) if self.password else ""
+        return (
+            f"postgresql://{self.user}:{password}@{self.host}:{self.port}/{self.name}"
+        )
+
+
+class RedisConfig(BaseSettings):
+    """Redis connection settings."""
+
+    model_config = SettingsConfigDict(env_prefix="REDIS_")
+
+    host: str = "localhost"
+    port: int = 6379
+    db: int = 0
+    password: str = ""
+
+    @property
+    def url(self) -> str:
+        if self.password:
+            return f"redis://:{self.password}@{self.host}:{self.port}/{self.db}"
+        return f"redis://{self.host}:{self.port}/{self.db}"
+
+
+class QRSizeConfig(BaseSettings):
+    """QR Code size presets (pixels)."""
+
+    model_config = SettingsConfigDict(env_prefix="QR_SIZE_")
+
+    small: int = 300
+    medium: int = 600
+    large: int = 1024
+
+    @field_validator("small", "medium", "large")
+    @classmethod
+    def validate_positive(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("QR size must be positive")
+        return v
+
+
+class WebhookConfig(BaseSettings):
+    """Webhook mode settings."""
 
-    # App URL for dynamic QR redirects
-    app_url = os.getenv("APP_URL", "https://go.make-qr.de")
-
-    # Site URL for fallback redirects (main website)
-    site_url = os.getenv("SITE_URL", "https://make-qr.de")
-
-    # Redirect server port (standalone mode)
-    redirect_port = int(os.getenv("REDIRECT_PORT", "8080"))
-
-    # QR Code size settings
-    qr_size_small = int(os.getenv("QR_SIZE_SMALL", "300"))
-    qr_size_medium = int(os.getenv("QR_SIZE_MEDIUM", "600"))
-    qr_size_large = int(os.getenv("QR_SIZE_LARGE", "1024"))
-
-    # Webhook configuration
-    bot_mode = os.getenv("BOT_MODE", "polling").lower()  # "polling" or "webhook"
-    webhook_host = os.getenv(
-        "WEBHOOK_HOST", "127.0.0.1"
-    )  # Default to localhost for security
-    webhook_port = int(os.getenv("WEBHOOK_PORT", "8080"))
-    webhook_path = os.getenv("WEBHOOK_PATH", "/webhook")
-    webhook_secret = os.getenv("WEBHOOK_SECRET", "")
-    webhook_base_url = os.getenv("WEBHOOK_BASE_URL", "")  # e.g., https://yourdomain.com
-    webhook_url = f"{webhook_base_url}{webhook_path}" if webhook_base_url else ""
-
-    return Config(
-        bot=BotConfig(token=bot_token, admin_ids=admin_ids),
-        locales_path=Path(__file__).parent.parent / "locales",
-        default_locale=os.getenv("DEFAULT_LOCALE", "en"),
-        app_url=app_url,
-        site_url=site_url,
-        redirect_port=redirect_port,
-        db_dsn=db_dsn,
-        database_url=db_dsn,  # Alias for compatibility
-        redis_url=redis_url,
-        event_bus_mode=event_bus_mode,
-        event_bus_consumer_name=event_bus_consumer_name,
-        event_store_enabled=event_store_enabled,
-        cache_warming_enabled=cache_warming_enabled,
-        qr_size_small=qr_size_small,
-        qr_size_medium=qr_size_medium,
-        qr_size_large=qr_size_large,
-        bot_mode=bot_mode,
-        webhook_host=webhook_host,
-        webhook_port=webhook_port,
-        webhook_path=webhook_path,
-        webhook_secret=webhook_secret,
-        webhook_url=webhook_url,
+    model_config = SettingsConfigDict(env_prefix="WEBHOOK_")
+
+    host: str = "127.0.0.1"
+    port: int = 8080
+    path: str = "/webhook"
+    secret: str = ""
+    base_url: str = ""
+
+    @property
+    def full_url(self) -> str:
+        return f"{self.base_url}{self.path}" if self.base_url else ""
+
+
+class EventBusConfig(BaseSettings):
+    """Event bus settings."""
+
+    model_config = SettingsConfigDict(env_prefix="EVENT_BUS_")
+
+    mode: str = "redis"
+    consumer_name: str = ""
+
+    @field_validator("mode")
+    @classmethod
+    def validate_mode(cls, v: str) -> str:
+        if v not in ("memory", "redis"):
+            raise ValueError(f"Invalid mode: {v}. Must be 'memory' or 'redis'")
+        return v
+
+
+class Config(BaseSettings):
+    """Main application configuration.
+
+    Loads from environment variables and .env file automatically.
+    All values are validated at startup — invalid config = immediate crash
+    with clear error messages.
+    """
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
     )
+
+    bot: BotConfig = Field(default_factory=BotConfig)
+    database: DatabaseConfig = Field(default_factory=DatabaseConfig)
+    redis: RedisConfig = Field(default_factory=RedisConfig)
+    qr_sizes: QRSizeConfig = Field(default_factory=QRSizeConfig)
+    webhook: WebhookConfig = Field(default_factory=WebhookConfig)
+    event_bus: EventBusConfig = Field(default_factory=EventBusConfig)
+
+    # Simple settings
+    app_url: str = Field(default="https://go.make-qr.de", alias="APP_URL")
+    site_url: str = Field(default="https://make-qr.de", alias="SITE_URL")
+    redirect_port: int = Field(default=8080, alias="REDIRECT_PORT")
+    default_locale: str = Field(default="en", alias="DEFAULT_LOCALE")
+    bot_mode: str = Field(default="polling", alias="BOT_MODE")
+    event_store_enabled: bool = Field(default=True, alias="EVENT_STORE_ENABLED")
+    cache_warming_enabled: bool = Field(default=True, alias="CACHE_WARMING_ENABLED")
+
+    @field_validator("bot_mode")
+    @classmethod
+    def validate_bot_mode(cls, v: str) -> str:
+        v = v.lower()
+        if v not in ("polling", "webhook"):
+            raise ValueError(f"Invalid bot_mode: {v}. Must be 'polling' or 'webhook'")
+        return v
+
+    @property
+    def locales_path(self) -> Path:
+        return Path(__file__).parent.parent / "locales"
+
+    @property
+    def db_dsn(self) -> str:
+        return self.database.dsn
+
+    @property
+    def redis_url(self) -> str:
+        return self.redis.url
+
+
+def load_config() -> Config:
+    """Load and validate configuration.
+
+    Raises pydantic.ValidationError with detailed messages
+    if any required field is missing or has invalid value.
+    """
+    return Config()

bot/di/modules/infrastructure.py

@@ -56,9 +56,9 @@ def register(self, container) -> None:
 
         # QR Code Generator with config-based sizes
         size_config = QRSizeConfig(
-            small=config.qr_size_small,
-            medium=config.qr_size_medium,
-            large=config.qr_size_large,
+            small=config.qr_sizes.small,
+            medium=config.qr_sizes.medium,
+            large=config.qr_sizes.large,
         )
         container.register_lazy(
             "qr_generator",
@@ -111,11 +111,11 @@ async def configure(self, container) -> None:
         except Exception as e:
             logger.warning(f"⚠️ Enhanced cache service initialization failed: {e}")
 
-        # Create and initialize event bus
+        # Event Bus configuration
         event_bus_instance = create_event_bus(
-            mode=config.event_bus_mode,
-            redis_url=config.redis_url if config.event_bus_mode == "redis" else None,
-            consumer_name=config.event_bus_consumer_name or None,
+            mode=config.event_bus.mode,
+            redis_url=config.redis_url if config.event_bus.mode == "redis" else None,
+            consumer_name=config.event_bus.consumer_name or None,
         )
 
         await initialize_event_bus(event_bus_instance)