51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
# policy.py (loader/utilities for user-supplied policy scripts)
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import inspect
|
|
import os
|
|
from types import ModuleType
|
|
from typing import Callable, Any
|
|
|
|
|
|
class PolicyLoadError(Exception):
|
|
pass
|
|
|
|
|
|
def _import_module_from_path(path: str) -> ModuleType:
|
|
if not os.path.exists(path):
|
|
raise PolicyLoadError(f"Policy file not found: {path}")
|
|
spec = importlib.util.spec_from_file_location("user_policy", path)
|
|
if spec is None or spec.loader is None:
|
|
raise PolicyLoadError("Could not load policy module spec.")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
try:
|
|
spec.loader.exec_module(mod) # type: ignore[reportAttributeAccessIssue]
|
|
except Exception as e:
|
|
raise PolicyLoadError(f"Error executing policy module: {e}") from e
|
|
return mod
|
|
|
|
def default_policy_path() -> str:
|
|
return os.path.join(os.path.dirname(__file__), "default_policy.py")
|
|
|
|
def load_policy(path: str) -> Callable[..., bool]:
|
|
"""
|
|
Load a user policy file and return the `decide` callable.
|
|
"""
|
|
mod = _import_module_from_path(path)
|
|
decide = getattr(mod, "decide", None)
|
|
if not callable(decide):
|
|
raise PolicyLoadError("Policy script must define a callable `decide(...)`.")
|
|
return decide
|
|
|
|
|
|
def call_policy(decide: Callable[..., Any], **kwargs: Any) -> Any:
|
|
"""
|
|
Call `decide` with only the parameters it accepts.
|
|
This lets us pass new optional context (like p, corr) without breaking old policies.
|
|
"""
|
|
sig = inspect.signature(decide)
|
|
accepted = {k: v for k, v in kwargs.items() if k in sig.parameters}
|
|
return decide(**accepted)
|
|
|