feat: add preset resolver
This commit is contained in:
@@ -73,10 +73,11 @@ class PipeRunner:
|
||||
pipe = self._parser.parse(pipeline)
|
||||
return self.run(pipe)
|
||||
|
||||
def check_conditions(self, conditions: List[str]) -> bool:
|
||||
for condition in conditions:
|
||||
if not self.resolve_value(condition):
|
||||
return False
|
||||
def check_conditions(self, conditions: Optional[List[str]]) -> bool:
|
||||
if conditions:
|
||||
for condition in conditions:
|
||||
if not self.resolve_value(condition):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
|
||||
@@ -1,37 +1,56 @@
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
from ...util import reflect
|
||||
from ..core import Pipe
|
||||
from ..util import reflect
|
||||
|
||||
__all__ = ("FilePipe",)
|
||||
__all__ = (
|
||||
"FileData",
|
||||
"FilePipe",
|
||||
)
|
||||
|
||||
|
||||
class FilePipe(Pipe[None, Any]):
|
||||
type = "file"
|
||||
class FileData:
|
||||
_LOADERS = {
|
||||
".toml": "tomli:load",
|
||||
".json": "json:load",
|
||||
}
|
||||
|
||||
def __init__(self, filename: Optional[str], loader: Optional[str] = None):
|
||||
self._filename = filename
|
||||
def __init__(self, filepath: Path, loader: Optional[str] = None):
|
||||
self._filepath = filepath
|
||||
self._loader = loader
|
||||
self._data = None
|
||||
|
||||
def _resolve_loader(self, filepath: Path):
|
||||
loader = self._loader or self._LOADERS.get(filepath.suffix)
|
||||
return reflect(loader) if loader else None
|
||||
|
||||
def _load_data(self):
|
||||
loader = self._resolve_loader(self._filepath)
|
||||
with open(self._filepath, "rb") as fd:
|
||||
if loader:
|
||||
self._data = loader(fd)
|
||||
else:
|
||||
self._data = fd.read()
|
||||
|
||||
def get(self, key: str):
|
||||
if self._data is None and self._filepath.exists():
|
||||
self._load_data()
|
||||
return self._data.get(key)
|
||||
|
||||
def __bool__(self):
|
||||
return self._filepath.exists()
|
||||
|
||||
|
||||
class FilePipe(Pipe[None, Any]):
|
||||
type = "file"
|
||||
|
||||
def __init__(self, filename: Optional[str], loader: Optional[str] = None):
|
||||
self._filename = filename
|
||||
self._loader = loader
|
||||
|
||||
def apply(self, value: Optional[str], basepath: Path) -> Optional[Any]:
|
||||
filepath = Path(value or self._filename) # TODO: raise none filepath exception
|
||||
if not filepath.is_absolute():
|
||||
filepath = basepath / filepath
|
||||
if not filepath.exists():
|
||||
return None
|
||||
loader = self._resolve_loader(filepath)
|
||||
with open(filepath, "rb") as fd:
|
||||
if loader:
|
||||
result = loader(fd)
|
||||
else:
|
||||
result = fd.read()
|
||||
return result
|
||||
return FileData(filepath, self._loader)
|
||||
|
||||
@@ -1,13 +0,0 @@
|
||||
import importlib
|
||||
from typing import Any
|
||||
|
||||
__all__ = ("reflect",)
|
||||
|
||||
|
||||
def reflect(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)
|
||||
Reference in New Issue
Block a user