89 lines
2.7 KiB
Python
89 lines
2.7 KiB
Python
import importlib
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, List, Optional
|
|
|
|
from .model import PipeStr
|
|
|
|
__all__ = ("ValueResolver",)
|
|
|
|
|
|
class ValueResolver:
|
|
PIPE_PATTERN = re.compile("^(\\w+)\\((.*)\\)")
|
|
FILE_LOADERS = {
|
|
".toml": "tomli:load",
|
|
".json": "json:load",
|
|
}
|
|
|
|
def __init__(self, basepath: Path):
|
|
self._basepath = basepath
|
|
|
|
def _reflect(self, value: str) -> Any:
|
|
if ":" in value:
|
|
module_name, item_name = value.split(":")
|
|
else:
|
|
module_name, item_name = "builtins", value
|
|
module = importlib.import_module(module_name)
|
|
return getattr(module, item_name)
|
|
|
|
def pipe_file(self, pipe, filename: str, loader: Optional[str] = None):
|
|
filepath = Path(filename)
|
|
if not filepath.is_absolute():
|
|
filepath = self._basepath / filepath
|
|
if loader is None:
|
|
loader = self.FILE_LOADERS.get(Path(filename).suffix)
|
|
if not filepath.exists():
|
|
return None
|
|
with open(filepath, "rb") as fd:
|
|
if loader:
|
|
loader_fun = self._reflect(loader)
|
|
data = loader_fun(fd)
|
|
else:
|
|
data = fd.read()
|
|
return data
|
|
|
|
def pipe_get(self, pipe, keypath: str):
|
|
path = keypath.split(".")
|
|
value = pipe.get(path[0])
|
|
for key in path[1:]:
|
|
if value:
|
|
value = value.get(key)
|
|
return value
|
|
|
|
def pipe_search(self, pipe, pattern: str, position_arg: str = "1"):
|
|
matched = re.compile(pattern).search(pipe)
|
|
position = int(position_arg)
|
|
if matched:
|
|
return matched.group(position)
|
|
|
|
def pipe_equal(self, pipe, value: Any):
|
|
return pipe == value
|
|
|
|
def pipe_convert(self, pipe, converter: str):
|
|
converter_fun = self._reflect(converter)
|
|
return converter_fun(pipe)
|
|
|
|
def pipe_not(self, pipe):
|
|
return not pipe
|
|
|
|
def resolve_value(self, operator: str) -> Any:
|
|
chain = [item.strip() for item in operator.split("|")]
|
|
pipe = None
|
|
for part in chain:
|
|
matched = self.PIPE_PATTERN.match(part)
|
|
if matched:
|
|
pipe_name = matched.group(1)
|
|
pipe_args = [item.strip() for item in matched.group(2).split(",")]
|
|
else:
|
|
pipe_name = part.strip()
|
|
pipe_args = []
|
|
pipe_func = getattr(self, f"pipe_{pipe_name}")
|
|
pipe = pipe_func(pipe, *pipe_args)
|
|
return pipe
|
|
|
|
def check_conditions(self, conditions: List[PipeStr]) -> bool:
|
|
for condition in conditions:
|
|
if not self.resolve_value(condition):
|
|
return False
|
|
return True
|