"""Typed application configuration. Values come from ``PHOTO_PIPELINE_*`` environment variables. Secrets use ``SecretStr`` so they are masked in logs, reprs, and model dumps, and are never returned by the API. Only ``.get_secret_value()`` exposes them, and only where a real external call needs them. pydantic-settings would do this too, but a prefix-scan over the declared fields is a few lines and one fewer dependency. Tuple-valued settings are lists in one variable: library roots are ``os.pathsep`` separated because they are paths, everything else is comma separated. """ from __future__ import annotations import os from pathlib import Path from typing import Mapping from pydantic import BaseModel, ConfigDict, SecretStr ENV_PREFIX = "PHOTO_PIPELINE_" ENV_FILE_VAR = f"{ENV_PREFIX}ENV_FILE" DEFAULT_ENV_FILE = Path(".env") COMMA_LIST_FIELDS = frozenset({"allowed_hosts", "trusted_proxies"}) # The archived CLI's variable names, so the configuration file an operator already # has keeps working. The vision provider reads the OpenAI SDK's names, and the # library root is configuration here rather than a bare path (US07-01 donor). LEGACY_ALIASES = { "LLM_API_KEY": "OPENAI_API_KEY", "GEMINI_API_KEY": "OPENAI_API_KEY", "LLM_BASE_URL": "OPENAI_BASE_URL", "LIBRARY": f"{ENV_PREFIX}LIBRARY_ROOTS", } def parse_env_file(text: str) -> dict[str, str]: """``KEY=value`` lines into a mapping. Comments, blanks, and quotes handled. Deliberately not a shell: no interpolation, no ``export``, no multi-line values. A configuration file that can run code is a configuration file that can be a vulnerability. """ values: dict[str, str] = {} for line in text.splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, raw = line.partition("=") key = key.strip() if not key or key.startswith("#"): continue value = raw.strip().strip('"').strip("'") values[key] = value alias = LEGACY_ALIASES.get(key) if alias: values.setdefault(alias, value) return values def load_env_file(path: Path | str | None = None) -> dict[str, str]: """Load ``PHOTO_PIPELINE_ENV_FILE`` (or ``./.env``) into the environment. Anything already exported wins: a file is the standing configuration, the shell is what you meant *this time*. Returns what it applied, which is what the CLI prints — names only, never values. """ candidate = path or os.environ.get(ENV_FILE_VAR) or DEFAULT_ENV_FILE candidate = Path(candidate) if not candidate.is_file(): return {} applied = {} for key, value in parse_env_file(candidate.read_text()).items(): if key not in os.environ: os.environ[key] = value applied[key] = value return applied class Config(BaseModel): model_config = ConfigDict(frozen=True) data_dir: Path = Path("data") db_path: Path | None = None # defaults to data_dir/photo_pipeline.db host: str = "127.0.0.1" port: int = 8000 log_level: str = "INFO" log_format: str = "json" # "json" or "text" # Trust boundary (US08-01). Empty means loopback only, which is what the app did # before there was a setting: a request whose Host is not a loopback name is # refused, and no secret is needed because nothing outside this machine can call. # Naming a real hostname here is what makes the app reachable through a reverse # proxy, and it is exactly then that ``access_secret`` becomes mandatory. allowed_hosts: tuple[str, ...] = () # Addresses whose ``X-Forwarded-Proto``/``X-Forwarded-Host`` may be believed. A # client that is not the proxy can otherwise declare its own origin. trusted_proxies: tuple[str, ...] = () # Exchanged for the session cookie at the bootstrap endpoint. Once set it is # required even on loopback, so a development setup cannot half-enable it. access_secret: SecretStr | None = None # Largest request body the API accepts. Every endpoint takes small JSON commands; # anything larger is a mistake or an attempt to exhaust memory (US07-02). max_request_bytes: int = 1_048_576 # Library boundary for path validation (os.pathsep-separated in the env var). library_roots: tuple[Path, ...] = () thumbnail_cache_quota_bytes: int = 500_000_000 thumbnail_max_pixels: int = 100_000_000 # Free space an archive destination must keep beyond the transfer itself. archive_free_space_reserve_bytes: int = 1_000_000_000 # Refuse every mutating request until a read-only dry run of the configured # library has been produced and explicitly approved (US07-07). Off by default so # a development setup is unchanged; turn it on before pointing the application at # a library whose photos cannot be replaced. require_dry_run_approval: bool = False vision_api_key: SecretStr | None = None immich_api_key: SecretStr | None = None immich_server_url: str = "" immich_go_binary: str = "immich-go" @property def database_path(self) -> Path: return self.db_path or (self.data_dir / "photo_pipeline.db") @property def database_url(self) -> str: return f"sqlite:///{self.database_path}" @property def thumbnail_cache_dir(self) -> Path: return self.data_dir / "cache" / "thumbs" @classmethod def from_env(cls, environ: Mapping[str, str] | None = None) -> "Config": env = os.environ if environ is None else environ if environ is None: load_env_file() # a file never overrides what the shell already set env = os.environ data: dict = {} for name in cls.model_fields: raw = env.get(ENV_PREFIX + name.upper()) if not raw: continue if name == "library_roots": data[name] = raw.split(os.pathsep) elif name in COMMA_LIST_FIELDS: data[name] = [part.strip() for part in raw.split(",") if part.strip()] else: data[name] = raw return cls(**data)