| """Console runtime settings, all from environment variables (12-factor: the |
| Docker compose file sets them; tests pass a dict).""" |
| from __future__ import annotations |
|
|
| import os |
| from collections.abc import Mapping |
| from dataclasses import dataclass |
| from pathlib import Path |
|
|
| from je_validation.ingest.snapshots import latest_snapshot |
|
|
|
|
| @dataclass(frozen=True) |
| class ConsoleSettings: |
| data_dir: Path |
| snapshot_dir: Path |
| openrouter_api_key: str |
| openrouter_base_url: str |
| max_concurrent_episodes: int |
| allow_unlimited_tokens: bool |
| port: int |
| space_host: str |
|
|
| @classmethod |
| def from_env(cls, env: Mapping[str, str] | None = None) -> ConsoleSettings: |
| e = os.environ if env is None else env |
| data_dir = Path(e.get("JE_DATA_DIR", "data")).resolve() |
| snap = e.get("JE_SNAPSHOT_DIR", "") |
| snapshot_dir = (Path(snap) if snap |
| else latest_snapshot(data_dir / "snapshots" / "ds_a")) |
| return cls( |
| data_dir=data_dir, |
| snapshot_dir=snapshot_dir, |
| openrouter_api_key=e.get("OPENROUTER_API_KEY", ""), |
| openrouter_base_url=e.get("JE_OPENROUTER_BASE_URL", |
| "https://openrouter.ai/api/v1"), |
| max_concurrent_episodes=int(e.get("JE_MAX_CONCURRENT_EPISODES", "5")), |
| allow_unlimited_tokens=e.get("JE_ALLOW_UNLIMITED_TOKENS", "") == "1", |
| port=int(e.get("JE_PORT", "8000")), |
| |
| space_host=e.get("SPACE_HOST", ""), |
| ) |
|
|