36 lines
1015 B
Python
36 lines
1015 B
Python
from __future__ import annotations
|
|
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
import re
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from schemas.automation import AutomationJobConfig
|
|
|
|
_CONFIG_NAME_PATTERN = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
|
|
|
|
|
|
def _automation_yaml_path(config_name: str) -> Path:
|
|
if not _CONFIG_NAME_PATTERN.fullmatch(config_name):
|
|
raise ValueError("invalid automation config name")
|
|
return (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "core"
|
|
/ "config"
|
|
/ "static"
|
|
/ "automation"
|
|
/ f"{config_name}.yaml"
|
|
)
|
|
|
|
|
|
@lru_cache(maxsize=16)
|
|
def load_static_automation_job_config(*, config_name: str) -> AutomationJobConfig:
|
|
path = _automation_yaml_path(config_name)
|
|
with path.open("r", encoding="utf-8") as file:
|
|
loaded: Any = yaml.safe_load(file) or {}
|
|
if not isinstance(loaded, dict):
|
|
raise ValueError(f"invalid automation config format: {path}")
|
|
return AutomationJobConfig.model_validate(loaded)
|