46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
|
|
DEFAULT_POLL_INTERVAL_SECONDS = 0.25
|
|
DEFAULT_WAIT_TIMEOUT_SECONDS = 10.0
|
|
DEFAULT_WORKFLOW_DB_PATH = "workflows/workflows.sqlite3"
|
|
|
|
POLL_INTERVAL_ENV = "WORKFLOW_POLL_INTERVAL_SECONDS"
|
|
WAIT_TIMEOUT_ENV = "WORKFLOW_WAIT_TIMEOUT_SECONDS"
|
|
DB_PATH_ENV = "WORKFLOW_DB_PATH"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WorkflowConfig:
|
|
poll_interval_seconds: float = DEFAULT_POLL_INTERVAL_SECONDS
|
|
wait_timeout_seconds: float = DEFAULT_WAIT_TIMEOUT_SECONDS
|
|
db_path: str = DEFAULT_WORKFLOW_DB_PATH
|
|
|
|
|
|
def load_config(env: Mapping[str, str] | None = None) -> WorkflowConfig:
|
|
values = env or os.environ
|
|
return WorkflowConfig(
|
|
poll_interval_seconds=_parse_float(
|
|
values.get(POLL_INTERVAL_ENV),
|
|
default=DEFAULT_POLL_INTERVAL_SECONDS,
|
|
),
|
|
wait_timeout_seconds=_parse_float(
|
|
values.get(WAIT_TIMEOUT_ENV),
|
|
default=DEFAULT_WAIT_TIMEOUT_SECONDS,
|
|
),
|
|
db_path=values.get(DB_PATH_ENV) or DEFAULT_WORKFLOW_DB_PATH,
|
|
)
|
|
|
|
|
|
def _parse_float(value: str | None, *, default: float) -> float:
|
|
if value is None:
|
|
return default
|
|
try:
|
|
parsed = float(value)
|
|
except ValueError:
|
|
return default
|
|
return parsed if parsed >= 0 else default
|