184 lines
5.4 KiB
Python
184 lines
5.4 KiB
Python
"""Render Compose + ``.env`` for a local stack profile."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from contextlib import suppress
|
|
from importlib.resources import files
|
|
from pathlib import Path
|
|
|
|
from honcho_cli.local.profile import LocalProfile
|
|
|
|
# Keys honcho start owns. Unknown lines in an existing .env are preserved.
|
|
MANAGED_KEYS = (
|
|
"AUTH_USE_AUTH",
|
|
"LOG_LEVEL",
|
|
"HONCHO_IMAGE",
|
|
"API_PORT",
|
|
"DB_PORT",
|
|
"REDIS_PORT",
|
|
)
|
|
|
|
# Host env forwarded into the profile .env (overrides config.toml).
|
|
_SETTINGS_PREFIXES = (
|
|
"LLM_",
|
|
"EMBEDDING_",
|
|
"DERIVER_",
|
|
"DIALECTIC_",
|
|
"DREAM_",
|
|
"SUMMARY_",
|
|
)
|
|
_LLM_KEYS = (
|
|
"LLM_OPENAI_API_KEY",
|
|
"LLM_ANTHROPIC_API_KEY",
|
|
"LLM_GEMINI_API_KEY",
|
|
)
|
|
|
|
_HEADER = (
|
|
"# Generated by honcho start. Extra keys below the managed block are preserved."
|
|
)
|
|
|
|
_PLACEHOLDERS = frozenset(
|
|
{
|
|
"",
|
|
"your-api-key-here",
|
|
"changeme",
|
|
"sk-...",
|
|
}
|
|
)
|
|
|
|
|
|
def is_placeholder_key(value: str | None) -> bool:
|
|
"""True when ``value`` is missing or a known template placeholder."""
|
|
if value is None:
|
|
return True
|
|
return value.strip() in _PLACEHOLDERS
|
|
|
|
|
|
def settings_from_environ() -> dict[str, str]:
|
|
"""Host env vars that map to Honcho settings. Empty/placeholder values skipped."""
|
|
return {
|
|
k: v
|
|
for k, v in os.environ.items()
|
|
if k.startswith(_SETTINGS_PREFIXES) and not is_placeholder_key(v)
|
|
}
|
|
|
|
|
|
def read_env_file(path: Path) -> dict[str, str]:
|
|
"""Parse a dotenv file into a dict. Last assignment of a key wins."""
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
lines = path.read_text(encoding="utf-8").splitlines()
|
|
except OSError:
|
|
return {}
|
|
out: dict[str, str] = {}
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#") or "=" not in stripped:
|
|
continue
|
|
k, _, v = stripped.partition("=")
|
|
out[k.strip()] = _unquote(v.strip())
|
|
return out
|
|
|
|
|
|
def read_env_value(path: Path, key: str) -> str | None:
|
|
"""Return the raw value for ``key`` in a dotenv file, or None."""
|
|
return read_env_file(path).get(key)
|
|
|
|
|
|
def has_provider_key(profile: LocalProfile, extra: dict[str, str]) -> bool:
|
|
"""True when host extra or profile ``.env`` has a real LLM API key."""
|
|
stored = {**read_env_file(profile.env_file()), **extra}
|
|
return any(not is_placeholder_key(stored.get(k)) for k in _LLM_KEYS)
|
|
|
|
|
|
def managed_env(profile: LocalProfile) -> dict[str, str]:
|
|
"""Values written into the managed block of ``.env``."""
|
|
return {
|
|
"AUTH_USE_AUTH": "false",
|
|
"LOG_LEVEL": "INFO",
|
|
"HONCHO_IMAGE": profile.image,
|
|
"API_PORT": str(profile.api_port),
|
|
"DB_PORT": str(profile.db_port),
|
|
"REDIS_PORT": str(profile.redis_port),
|
|
}
|
|
|
|
|
|
def upsert_env(
|
|
path: Path,
|
|
updates: dict[str, str],
|
|
*,
|
|
drop: tuple[str, ...] = (),
|
|
) -> None:
|
|
"""Write ``updates``, preserving unrelated user lines.
|
|
|
|
Managed keys are written first (stable order), then any other keys in
|
|
``updates``. Keys in ``drop`` are removed and not rewritten. Drops a
|
|
previous generated header so it is not duplicated.
|
|
"""
|
|
drop_set = frozenset(drop)
|
|
extras: list[str] = []
|
|
if path.exists():
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
stripped = line.strip()
|
|
if stripped == _HEADER or stripped.startswith(
|
|
"# Generated by honcho start"
|
|
):
|
|
continue
|
|
if not stripped or stripped.startswith("#"):
|
|
extras.append(line)
|
|
continue
|
|
if "=" in stripped:
|
|
k, _, _ = stripped.partition("=")
|
|
name = k.strip()
|
|
if name in updates or name in MANAGED_KEYS or name in drop_set:
|
|
continue
|
|
extras.append(line)
|
|
|
|
managed = [f"{k}={updates[k]}" for k in MANAGED_KEYS if k in updates]
|
|
extra_updates = [
|
|
f"{k}={v}" for k, v in updates.items() if k not in MANAGED_KEYS
|
|
]
|
|
body = [_HEADER, *managed]
|
|
if extra_updates:
|
|
if not body[-1].startswith("#"):
|
|
body.append("")
|
|
body.extend(extra_updates)
|
|
if extras:
|
|
# Keep a blank line between generated and user keys when there are extras.
|
|
if extras[0].strip():
|
|
body.append("")
|
|
body.extend(extras)
|
|
path.write_text("\n".join(body) + "\n")
|
|
with suppress(OSError):
|
|
os.chmod(path, 0o600)
|
|
|
|
|
|
def render_stack(
|
|
profile: LocalProfile,
|
|
extra: dict[str, str] | None = None,
|
|
drop: tuple[str, ...] = (),
|
|
) -> None:
|
|
"""Write compose, init.sql, and .env into the profile directory."""
|
|
directory = profile.dir()
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
with suppress(OSError):
|
|
os.chmod(directory, 0o700)
|
|
|
|
templates = files("honcho_cli.local.templates")
|
|
compose = templates.joinpath("docker-compose.yml").read_text(encoding="utf-8")
|
|
init_sql = templates.joinpath("init.sql").read_text(encoding="utf-8")
|
|
profile.compose_file().write_text(compose)
|
|
(directory / "init.sql").write_text(init_sql)
|
|
updates = managed_env(profile)
|
|
if extra:
|
|
updates.update(extra)
|
|
upsert_env(profile.env_file(), updates, drop=drop)
|
|
|
|
|
|
def _unquote(value: str) -> str:
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}:
|
|
return value[1:-1]
|
|
return value
|