86 lines
3.6 KiB
Python
86 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass
|
|
|
|
|
|
def _env_bool(name: str, default: bool) -> bool:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _env_set(name: str) -> frozenset[str]:
|
|
return frozenset(value.strip() for value in os.getenv(name, "").split(",") if value.strip())
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RuntimeSettings:
|
|
environment: str = "development"
|
|
backend: str = "memory"
|
|
allow_ephemeral: bool = True
|
|
api_host: str = "127.0.0.1"
|
|
api_port: int = 8010
|
|
api_base: str = "http://127.0.0.1:8010"
|
|
mcp_host: str = "127.0.0.1"
|
|
mcp_port: int = 8011
|
|
mcp_transport: str = "http"
|
|
database_url: str | None = None
|
|
api_token: str | None = None
|
|
v2_groups: frozenset[str] = frozenset()
|
|
v2_write_groups: frozenset[str] = frozenset()
|
|
legacy_write_groups: frozenset[str] = frozenset()
|
|
legacy_health: bool = False
|
|
|
|
def __post_init__(self) -> None:
|
|
overlap = self.v2_write_groups & self.legacy_write_groups
|
|
if overlap:
|
|
joined = ", ".join(sorted(overlap))
|
|
raise ValueError(f"hub-core and Core Hub write groups overlap: {joined}")
|
|
unknown = self.v2_write_groups - self.v2_groups
|
|
if unknown:
|
|
joined = ", ".join(sorted(unknown))
|
|
raise ValueError(f"hub-core write groups are not enabled: {joined}")
|
|
|
|
@classmethod
|
|
def from_env(cls) -> RuntimeSettings:
|
|
environment = os.getenv("HUB_CORE_ENV", "development")
|
|
default_ephemeral = environment in {"development", "test"}
|
|
api_host = os.getenv("HUB_CORE_API_HOST", "127.0.0.1")
|
|
api_port = int(os.getenv("HUB_CORE_API_PORT", "8010"))
|
|
return cls(
|
|
environment=environment,
|
|
backend=os.getenv("HUB_CORE_BACKEND", "memory"),
|
|
allow_ephemeral=_env_bool("HUB_CORE_ALLOW_EPHEMERAL", default_ephemeral),
|
|
api_host=api_host,
|
|
api_port=api_port,
|
|
api_base=os.getenv("HUB_CORE_API_BASE", f"http://127.0.0.1:{api_port}"),
|
|
mcp_host=os.getenv("HUB_CORE_MCP_HOST", "127.0.0.1"),
|
|
mcp_port=int(os.getenv("HUB_CORE_MCP_PORT", "8011")),
|
|
mcp_transport=os.getenv("HUB_CORE_MCP_TRANSPORT", "http"),
|
|
database_url=os.getenv("HUB_CORE_DATABASE_URL") or os.getenv("DATABASE_URL"),
|
|
api_token=os.getenv("HUB_CORE_API_TOKEN") or os.getenv("CORE_HUB_API_TOKEN"),
|
|
v2_groups=_env_set("HUB_CORE_V2_GROUPS"),
|
|
v2_write_groups=_env_set("HUB_CORE_V2_WRITE_GROUPS"),
|
|
legacy_write_groups=_env_set("CORE_HUB_V2_WRITE_GROUPS"),
|
|
legacy_health=_env_bool("HUB_CORE_LEGACY_HEALTH", False),
|
|
)
|
|
|
|
def readiness_checks(self, store_backend: str) -> dict[str, str]:
|
|
ephemeral_allowed = store_backend != "memory" or self.allow_ephemeral
|
|
return {
|
|
"environment": self.environment,
|
|
"configured_backend": self.backend,
|
|
"active_backend": store_backend,
|
|
"ephemeral_backend": "allowed" if ephemeral_allowed else "not_allowed",
|
|
"contract": "helixforge.hub-extension/0.1.0",
|
|
"v2_groups": ",".join(sorted(self.v2_groups)) or "none",
|
|
"v2_write_groups": ",".join(sorted(self.v2_write_groups)) or "none",
|
|
"legacy_write_groups": ",".join(sorted(self.legacy_write_groups)) or "none",
|
|
}
|
|
|
|
def is_ready(self, store_backend: str) -> bool:
|
|
return self.backend == store_backend and (
|
|
store_backend != "memory" or self.allow_ephemeral
|
|
)
|