diff --git a/src/wic/api/pythonapi.py b/src/wic/api/pythonapi.py index 3dc9de54..8667c644 100644 --- a/src/wic/api/pythonapi.py +++ b/src/wic/api/pythonapi.py @@ -1,3 +1,4 @@ +# pylint: disable=W1203 """CLT utilities.""" import logging import subprocess @@ -17,12 +18,17 @@ from wic.api._types import CWL_TYPES_DICT if PYDANTIC_V2: - from pydantic import ConfigDict, field_validator + from pydantic import ( # type: ignore # pylint: disable=E0611, C0412 + ConfigDict, + field_validator, + ) else: from pydantic import validator logger = logging.getLogger("WIC Python API") +_WIC_PATH = Path(__file__).parent.parent.parent.parent # WIC dir + class InvalidInputValueError(Exception): pass @@ -330,16 +336,18 @@ def _save_cwl(self, path: Path) -> None: file.write(yaml.dump(self.yaml)) -@pydantic_dataclass +if PYDANTIC_V2: + DATACLASS_CONFIG = ConfigDict(validate_assignment=True) +else: + # mypy marks this incorrect if Pydantic V2 + DATACLASS_CONFIG = ConfigDict(validate_on_init=True, validate_assignment=True) # type: ignore + + +@pydantic_dataclass(config=DATACLASS_CONFIG) class Workflow: steps: list[Step] name: str - path: Path - if PYDANTIC_V2: - yml_path: Path = field(init=False, repr=False) # pylint: disable=E3701 - else: - # mypy incorrectly marks this as redef - yml_path: Optional[Path] = field(init=False, default=None, repr=False) # type: ignore # pylint: disable=E3701 + yml_path: Optional[Path] = field(default=None, init=False, repr=False) # pylint: disable=E3701 def __post_init__(self) -> None: for s in self.steps: @@ -350,8 +358,10 @@ def __post_init__(self) -> None: f"{s.clt_name} is missing required inputs" ) from exc - def step(self, step_: Step) -> None: + def append(self, step_: Step) -> None: """Append step to Workflow.""" + if not isinstance(step_, Step): + raise TypeError("step must be a Step") self.steps.append(step_) @property @@ -361,8 +371,8 @@ def yaml(self) -> dict[str, Any]: return d def _save_yaml(self) -> None: - self.path.mkdir(parents=True, exist_ok=True) - self.yml_path = self.path.joinpath(f"{self.name}.yml") + _WIC_PATH.mkdir(parents=True, exist_ok=True) + self.yml_path = _WIC_PATH.joinpath(f"{self.name}.yml") with open(self.yml_path, "w", encoding="utf-8") as file: file.write(yaml.dump(self.yaml)) @@ -371,15 +381,19 @@ def _save_all_cwl(self) -> None: This is necessary for WIC to compile the workflow. """ - self.path.mkdir(parents=True, exist_ok=True) + _WIC_PATH.mkdir(parents=True, exist_ok=True) for s in self.steps: try: - s._save_cwl(self.path) # pylint: disable=W0212 + s._save_cwl(_WIC_PATH) # pylint: disable=W0212 except BaseException as exc: raise exc def compile(self, run_local: bool = False) -> Path: - """Compile Workflow using WIC.""" + """Compile Workflow using WIC. + + Returns path to compiled CWL Workflow. + """ + self._save_all_cwl() self._save_yaml() logger.info(f"Compiling {self.name}") @@ -390,7 +404,7 @@ def compile(self, run_local: bool = False) -> Path: proc = subprocess.run( args=args, capture_output=True, - cwd=self.path, + cwd=_WIC_PATH, check=True, text=True, universal_newlines=True, @@ -400,7 +414,7 @@ def compile(self, run_local: bool = False) -> Path: logger.info(exc.stdout) raise exc logger.info(proc.stdout) - return self.path.joinpath("autogenerated", f"{self.name}.cwl") + return _WIC_PATH.joinpath("autogenerated", f"{self.name}.cwl") def run(self) -> None: """Run compiled workflow."""