129 lines
3.7 KiB
Python
129 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from typing import Literal
|
|
|
|
|
|
EnvironmentName = Literal["local", "test", "production"]
|
|
SUPPORTED_DATABASE_PREFIXES = (
|
|
"sqlite:///",
|
|
"postgresql://",
|
|
"postgresql+psycopg://",
|
|
)
|
|
|
|
|
|
class CloudConfigurationError(ValueError):
|
|
"""Raised when control-plane configuration is unsafe or invalid."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CloudControlConfig:
|
|
environment: EnvironmentName = "local"
|
|
database_url: str = "sqlite:///cloud/cloud.sqlite3"
|
|
scheduler_interval_seconds: float = 1.0
|
|
lease_reaper_interval_seconds: float = 5.0
|
|
lease_duration_seconds: float = 60.0
|
|
max_task_attempts: int = 3
|
|
allow_insecure_anonymous: bool = False
|
|
|
|
|
|
def load_control_config(
|
|
env: Mapping[str, str] | None = None,
|
|
) -> CloudControlConfig:
|
|
values = os.environ if env is None else env
|
|
environment = values.get("CLOUD_ENVIRONMENT", "local").strip().lower()
|
|
if environment not in {"local", "test", "production"}:
|
|
raise CloudConfigurationError(
|
|
"CLOUD_ENVIRONMENT must be local, test, or production"
|
|
)
|
|
|
|
database_url = values.get(
|
|
"CLOUD_DATABASE_URL",
|
|
"sqlite:///cloud/cloud.sqlite3",
|
|
).strip()
|
|
if not database_url.startswith(SUPPORTED_DATABASE_PREFIXES):
|
|
raise CloudConfigurationError(
|
|
"CLOUD_DATABASE_URL must use sqlite or postgresql"
|
|
)
|
|
|
|
config = CloudControlConfig(
|
|
environment=environment, # type: ignore[arg-type]
|
|
database_url=database_url,
|
|
scheduler_interval_seconds=_positive_float(
|
|
values,
|
|
"CLOUD_SCHEDULER_INTERVAL_SECONDS",
|
|
1.0,
|
|
),
|
|
lease_reaper_interval_seconds=_positive_float(
|
|
values,
|
|
"CLOUD_LEASE_REAPER_INTERVAL_SECONDS",
|
|
5.0,
|
|
),
|
|
lease_duration_seconds=_positive_float(
|
|
values,
|
|
"CLOUD_LEASE_DURATION_SECONDS",
|
|
60.0,
|
|
),
|
|
max_task_attempts=_positive_int(
|
|
values,
|
|
"CLOUD_MAX_TASK_ATTEMPTS",
|
|
3,
|
|
),
|
|
allow_insecure_anonymous=_parse_bool(
|
|
values.get("CLOUD_ALLOW_INSECURE_ANONYMOUS"),
|
|
default=False,
|
|
),
|
|
)
|
|
if config.environment == "production" and config.allow_insecure_anonymous:
|
|
raise CloudConfigurationError(
|
|
"anonymous access cannot be enabled in production"
|
|
)
|
|
return config
|
|
|
|
|
|
def _positive_float(
|
|
values: Mapping[str, str],
|
|
name: str,
|
|
default: float,
|
|
) -> float:
|
|
raw_value = values.get(name)
|
|
if raw_value is None:
|
|
return default
|
|
try:
|
|
value = float(raw_value)
|
|
except ValueError as exc:
|
|
raise CloudConfigurationError(f"{name} must be a number") from exc
|
|
if value <= 0:
|
|
raise CloudConfigurationError(f"{name} must be greater than zero")
|
|
return value
|
|
|
|
|
|
def _positive_int(
|
|
values: Mapping[str, str],
|
|
name: str,
|
|
default: int,
|
|
) -> int:
|
|
raw_value = values.get(name)
|
|
if raw_value is None:
|
|
return default
|
|
try:
|
|
value = int(raw_value)
|
|
except ValueError as exc:
|
|
raise CloudConfigurationError(f"{name} must be an integer") from exc
|
|
if value <= 0:
|
|
raise CloudConfigurationError(f"{name} must be greater than zero")
|
|
return value
|
|
|
|
|
|
def _parse_bool(value: str | None, *, default: bool) -> bool:
|
|
if value is None:
|
|
return default
|
|
normalized = value.strip().lower()
|
|
if normalized in {"1", "true", "yes", "on", "enabled"}:
|
|
return True
|
|
if normalized in {"0", "false", "no", "off", "disabled", ""}:
|
|
return False
|
|
raise CloudConfigurationError("boolean configuration value is invalid")
|