import os from typing import Optional from urllib.parse import urlsplit, urlunsplit from pydantic import model_validator from pydantic_settings import BaseSettings, SettingsConfigDict _DOCKER_SERVICE_ALIASES = {"postgres", "redis", "minio"} def _is_running_in_container() -> bool: if os.path.exists("/.dockerenv"): return True try: with open("/proc/1/cgroup", "r", encoding="utf-8") as handle: return "docker" in handle.read() or "containerd" in handle.read() except OSError: return False def _normalize_service_host(host: str) -> str: if _is_running_in_container(): return host return "localhost" if host in _DOCKER_SERVICE_ALIASES else host def _normalize_service_url(url: str) -> str: parsed = urlsplit(url) if not parsed.hostname: return url normalized_host = _normalize_service_host(parsed.hostname) if normalized_host == parsed.hostname: return url netloc = normalized_host if parsed.username: auth = parsed.username if parsed.password: auth = f"{auth}:{parsed.password}" netloc = f"{auth}@{netloc}" if parsed.port: netloc = f"{netloc}:{parsed.port}" return urlunsplit((parsed.scheme, netloc, parsed.path, parsed.query, parsed.fragment)) class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=".env", case_sensitive=False, extra="ignore", ) # Database postgres_db: str = "hartomat" postgres_user: str = "hartomat" postgres_password: str = "hartomat" postgres_host: str = "localhost" postgres_port: int = 5432 @property def database_url(self) -> str: return ( f"postgresql+asyncpg://{self.postgres_user}:{self.postgres_password}" f"@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}" ) @property def database_url_sync(self) -> str: return ( f"postgresql://{self.postgres_user}:{self.postgres_password}" f"@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}" ) # Redis / Celery redis_url: str = "redis://localhost:6379/0" # Queue for shadow-mode workflow renders (second GPU worker). workflow_shadow_render_queue: str = "asset_pipeline_light" # When non-empty AND that queue has active workers, still renders from the # legacy dispatch path are routed here instead of asset_pipeline, enabling # concurrent still rendering on multi-GPU setups. # Set MULTI_GPU_LIGHT_RENDER_QUEUE=asset_pipeline_light in docker-compose # for the render-worker-light service and restart the workers. multi_gpu_light_render_queue: str = "" @model_validator(mode="after") def normalize_runtime_hosts(self) -> "Settings": self.postgres_host = _normalize_service_host(self.postgres_host) self.redis_url = _normalize_service_url(self.redis_url) return self @model_validator(mode="after") def reject_insecure_jwt_secret_in_production(self) -> "Settings": if self.jwt_secret_key == "changeme" and _is_running_in_container(): raise ValueError( "JWT_SECRET_KEY must be set to a secure random value in production. " "The default 'changeme' key is not permitted when running inside a container." ) return self # JWT jwt_secret_key: str = "changeme" jwt_algorithm: str = "HS256" jwt_access_token_expire_minutes: int = 480 # Azure OpenAI azure_openai_api_key: Optional[str] = None azure_openai_endpoint: Optional[str] = None azure_openai_deployment: str = "gpt-4o" azure_openai_api_version: str = "2024-02-01" # CORS (set CORS_ORIGINS='["https://app.example.com"]' in production) cors_origins: list[str] = [ "http://localhost:5173", "http://localhost:3000", "http://frontend:5173", "http://localhost:8888", ] # Order numbering (set ORDER_NUMBER_PREFIX to change the "SA-" prefix for white-labelling) order_number_prefix: str = "SA" # Internal API (used by chat_service for self-calls — override in Docker if port changes) internal_api_base_url: str = "http://localhost:8888" # File Storage upload_dir: str = "/app/uploads" max_upload_size_mb: int = 500 settings = Settings()