[ENH]: Update QueueContextAdapter interface for backward compatibility #503
6 changed files with 59 additions and 52 deletions
1
docs/changes/newsfragments/503.enh
Normal file
1
docs/changes/newsfragments/503.enh
Normal file
|
|
@ -0,0 +1 @@
|
|||
Revert ``pre_run`` and ``pre_collect`` parameters for queue context adapters by `Synchon Mandal`_
|
||||
|
|
@ -35,9 +35,9 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
|
|||
The path to the YAML config file.
|
||||
elements : ``Elements``
|
||||
Element(s) to process. Will be used to index the DataGrabber.
|
||||
pre_run_cmds : str or None, optional
|
||||
pre_run : str or None, optional
|
||||
Extra shell commands to source before the run (default None).
|
||||
pre_collect_cmds : str or None, optional
|
||||
pre_collect : str or None, optional
|
||||
Extra shell commands to source before the collect (default None).
|
||||
env : :class:`.QueueContextEnv` or None, optional
|
||||
The environment configuration. If None, will run without a
|
||||
|
|
@ -63,8 +63,8 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
|
|||
job_dir: Path
|
||||
yaml_config_path: Path
|
||||
elements: Elements
|
||||
pre_run_cmds: str | None = None
|
||||
pre_collect_cmds: str | None = None
|
||||
pre_run: str | None = None
|
||||
pre_collect: str | None = None
|
||||
env: QueueContextEnv | None = None
|
||||
verbose: str = "info"
|
||||
verbose_datalad: str | None = None
|
||||
|
|
@ -105,7 +105,7 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
|
|||
|
||||
return "\n".join(elements_to_run)
|
||||
|
||||
def pre_run(self) -> str:
|
||||
def pre_run_cmds(self) -> str:
|
||||
"""Return pre-run commands."""
|
||||
fixed = (
|
||||
f"#!/usr/bin/env {self.env['shell']}\n\n"
|
||||
|
|
@ -113,10 +113,10 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
|
|||
"# Force datalad to run in non-interactive mode\n"
|
||||
"DATALAD_UI_INTERACTIVE=false\n"
|
||||
)
|
||||
var = self.pre_run_cmds or ""
|
||||
var = self.pre_run or ""
|
||||
return fixed + "\n" + var
|
||||
|
||||
def run(self) -> str:
|
||||
def run_cmds(self) -> str:
|
||||
"""Return run commands."""
|
||||
verbose_args = f"--verbose {self.verbose}"
|
||||
if self.verbose_datalad:
|
||||
|
|
@ -141,16 +141,16 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
|
|||
f"--element"
|
||||
)
|
||||
|
||||
def pre_collect(self) -> str:
|
||||
def pre_collect_cmds(self) -> str:
|
||||
"""Return pre-collect commands."""
|
||||
fixed = (
|
||||
f"#!/usr/bin/env {self.env['shell']}\n\n"
|
||||
"# This script is auto-generated by junifer.\n"
|
||||
)
|
||||
var = self.pre_collect_cmds or ""
|
||||
var = self.pre_collect or ""
|
||||
return fixed + "\n" + var
|
||||
|
||||
def collect(self) -> str:
|
||||
def collect_cmds(self) -> str:
|
||||
"""Return collect commands."""
|
||||
verbose_args = f"--verbose {self.verbose}"
|
||||
if self.verbose_datalad:
|
||||
|
|
@ -196,14 +196,14 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
|
|||
f"Writing {self._pre_run_path.name} to {self.job_dir.resolve()!s}"
|
||||
)
|
||||
self._pre_run_path.touch()
|
||||
self._pre_run_path.write_text(textwrap.dedent(self.pre_run()))
|
||||
self._pre_run_path.write_text(textwrap.dedent(self.pre_run_cmds()))
|
||||
make_executable(self._pre_run_path)
|
||||
# Create run
|
||||
logger.info(
|
||||
f"Writing {self._run_path.name} to {self.job_dir.resolve()!s}"
|
||||
)
|
||||
self._run_path.touch()
|
||||
self._run_path.write_text(textwrap.dedent(self.run()))
|
||||
self._run_path.write_text(textwrap.dedent(self.run_cmds()))
|
||||
make_executable(self._run_path)
|
||||
# Create pre collect
|
||||
logger.info(
|
||||
|
|
@ -211,14 +211,16 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
|
|||
f"{self.job_dir.resolve()!s}"
|
||||
)
|
||||
self._pre_collect_path.touch()
|
||||
self._pre_collect_path.write_text(textwrap.dedent(self.pre_collect()))
|
||||
self._pre_collect_path.write_text(
|
||||
textwrap.dedent(self.pre_collect_cmds())
|
||||
)
|
||||
make_executable(self._pre_collect_path)
|
||||
# Create collect
|
||||
logger.info(
|
||||
f"Writing {self._collect_path.name} to {self.job_dir.resolve()!s}"
|
||||
)
|
||||
self._collect_path.touch()
|
||||
self._collect_path.write_text(textwrap.dedent(self.collect()))
|
||||
self._collect_path.write_text(textwrap.dedent(self.collect_cmds()))
|
||||
make_executable(self._collect_path)
|
||||
# Submit if required
|
||||
run_cmd = f"sh {self._run_path.resolve()!s}"
|
||||
|
|
|
|||
|
|
@ -52,9 +52,9 @@ class HTCondorAdapter(QueueContextAdapter):
|
|||
The path to the YAML config file.
|
||||
elements : ``Elements``
|
||||
Element(s) to process. Will be used to index the DataGrabber.
|
||||
pre_run_cmds : str or None, optional
|
||||
pre_run : str or None, optional
|
||||
Extra shell commands to source before the run (default None).
|
||||
pre_collect_cmds : str or None, optional
|
||||
pre_collect : str or None, optional
|
||||
Extra shell commands to source before the collect (default None).
|
||||
env : :class:`.QueueContextEnv` or None, optional
|
||||
The environment configuration. If None, will run without a
|
||||
|
|
@ -91,8 +91,8 @@ class HTCondorAdapter(QueueContextAdapter):
|
|||
job_dir: Path
|
||||
yaml_config_path: Path
|
||||
elements: Elements
|
||||
pre_run_cmds: str | None = None
|
||||
pre_collect_cmds: str | None = None
|
||||
pre_run: str | None = None
|
||||
pre_collect: str | None = None
|
||||
env: QueueContextEnv | None = None
|
||||
verbose: str = "info"
|
||||
verbose_datalad: str | None = None
|
||||
|
|
@ -127,7 +127,7 @@ class HTCondorAdapter(QueueContextAdapter):
|
|||
)
|
||||
self._dag_path = self.job_dir / f"{self.job_name}.dag"
|
||||
|
||||
def pre_run(self) -> str:
|
||||
def pre_run_cmds(self) -> str:
|
||||
"""Return pre-run commands."""
|
||||
fixed = (
|
||||
f"#!/usr/bin/env {self.env['shell']}\n\n"
|
||||
|
|
@ -135,10 +135,10 @@ class HTCondorAdapter(QueueContextAdapter):
|
|||
"# Force datalad to run in non-interactive mode\n"
|
||||
"DATALAD_UI_INTERACTIVE=false\n"
|
||||
)
|
||||
var = self.pre_run_cmds or ""
|
||||
var = self.pre_run or ""
|
||||
return fixed + "\n" + var
|
||||
|
||||
def run(self) -> str:
|
||||
def run_cmds(self) -> str:
|
||||
"""Return run commands."""
|
||||
verbose_args = f"--verbose {self.verbose} "
|
||||
if self.verbose_datalad is not None:
|
||||
|
|
@ -176,19 +176,19 @@ class HTCondorAdapter(QueueContextAdapter):
|
|||
var = self.extra_preamble or ""
|
||||
return fixed + "\n" + var + "\n" + "queue"
|
||||
|
||||
def pre_collect(self) -> str:
|
||||
def pre_collect_cmds(self) -> str:
|
||||
"""Return pre-collect commands."""
|
||||
fixed = (
|
||||
f"#!/usr/bin/env {self.env['shell']}\n\n"
|
||||
"# This script is auto-generated by junifer.\n"
|
||||
)
|
||||
var = self.pre_collect_cmds or ""
|
||||
var = self.pre_collect or ""
|
||||
# Add commands if collect="yes"
|
||||
if self.collect_task == "yes":
|
||||
var += 'if [ "${1}" == "4" ]; then\n exit 1\nfi\n'
|
||||
return fixed + "\n" + var
|
||||
|
||||
def collect(self) -> str:
|
||||
def collect_cmds(self) -> str:
|
||||
"""Return collect commands."""
|
||||
verbose_args = f"--verbose {self.verbose} "
|
||||
if self.verbose_datalad is not None:
|
||||
|
|
@ -222,7 +222,7 @@ class HTCondorAdapter(QueueContextAdapter):
|
|||
var = self.extra_preamble or ""
|
||||
return fixed + "\n" + var + "\n" + "queue"
|
||||
|
||||
def dag(self) -> str:
|
||||
def dag_cmds(self) -> str:
|
||||
"""Return HTCondor DAG commands."""
|
||||
fixed = ""
|
||||
for idx, element in enumerate(self.elements):
|
||||
|
|
@ -277,7 +277,7 @@ class HTCondorAdapter(QueueContextAdapter):
|
|||
f"Writing {self._pre_run_path.name} to {self.job_dir.resolve()!s}"
|
||||
)
|
||||
self._pre_run_path.touch()
|
||||
self._pre_run_path.write_text(textwrap.dedent(self.pre_run()))
|
||||
self._pre_run_path.write_text(textwrap.dedent(self.pre_run_cmds()))
|
||||
make_executable(self._pre_run_path)
|
||||
# Create run
|
||||
logger.debug(
|
||||
|
|
@ -285,14 +285,16 @@ class HTCondorAdapter(QueueContextAdapter):
|
|||
f"{self.job_dir.resolve()!s}"
|
||||
)
|
||||
self._submit_run_path.touch()
|
||||
self._submit_run_path.write_text(textwrap.dedent(self.run()))
|
||||
self._submit_run_path.write_text(textwrap.dedent(self.run_cmds()))
|
||||
# Create pre collect
|
||||
logger.info(
|
||||
f"Writing {self._pre_collect_path.name} to "
|
||||
f"{self.job_dir.resolve()!s}"
|
||||
)
|
||||
self._pre_collect_path.touch()
|
||||
self._pre_collect_path.write_text(textwrap.dedent(self.pre_collect()))
|
||||
self._pre_collect_path.write_text(
|
||||
textwrap.dedent(self.pre_collect_cmds())
|
||||
)
|
||||
make_executable(self._pre_collect_path)
|
||||
# Create collect
|
||||
logger.debug(
|
||||
|
|
@ -300,13 +302,15 @@ class HTCondorAdapter(QueueContextAdapter):
|
|||
f"{self.job_dir.resolve()!s}"
|
||||
)
|
||||
self._submit_collect_path.touch()
|
||||
self._submit_collect_path.write_text(textwrap.dedent(self.collect()))
|
||||
self._submit_collect_path.write_text(
|
||||
textwrap.dedent(self.collect_cmds())
|
||||
)
|
||||
# Create DAG
|
||||
logger.debug(
|
||||
f"Writing {self._dag_path.name} to {self.job_dir.resolve()!s}"
|
||||
)
|
||||
self._dag_path.touch()
|
||||
self._dag_path.write_text(textwrap.dedent(self.dag()))
|
||||
self._dag_path.write_text(textwrap.dedent(self.dag_cmds()))
|
||||
# Submit if required
|
||||
condor_submit_dag_cmd = [
|
||||
"condor_submit_dag",
|
||||
|
|
|
|||
|
|
@ -64,11 +64,11 @@ class QueueContextAdapter(BaseModel, ABC):
|
|||
|
||||
model_config = ConfigDict(
|
||||
use_enum_values=True,
|
||||
extra="allow",
|
||||
extra="forbid",
|
||||
)
|
||||
|
||||
@abstractmethod
|
||||
def pre_run(self) -> str:
|
||||
def pre_run_cmds(self) -> str:
|
||||
"""Return pre-run commands."""
|
||||
raise_error(
|
||||
msg="Concrete classes need to implement pre_run()",
|
||||
|
|
@ -76,7 +76,7 @@ class QueueContextAdapter(BaseModel, ABC):
|
|||
)
|
||||
|
||||
@abstractmethod
|
||||
def run(self) -> str:
|
||||
def run_cmds(self) -> str:
|
||||
"""Return run commands."""
|
||||
raise_error(
|
||||
msg="Concrete classes need to implement run()",
|
||||
|
|
@ -84,7 +84,7 @@ class QueueContextAdapter(BaseModel, ABC):
|
|||
)
|
||||
|
||||
@abstractmethod
|
||||
def pre_collect(self) -> str:
|
||||
def pre_collect_cmds(self) -> str:
|
||||
"""Return pre-collect commands."""
|
||||
raise_error(
|
||||
msg="Concrete classes need to implement pre_collect()",
|
||||
|
|
@ -92,7 +92,7 @@ class QueueContextAdapter(BaseModel, ABC):
|
|||
)
|
||||
|
||||
@abstractmethod
|
||||
def collect(self) -> str:
|
||||
def collect_cmds(self) -> str:
|
||||
"""Return collect commands."""
|
||||
raise_error(
|
||||
msg="Concrete classes need to implement collect()",
|
||||
|
|
|
|||
|
|
@ -73,10 +73,10 @@ def test_GnuParallelLocalAdapter_pre_run(
|
|||
yaml_config_path=Path("."),
|
||||
elements=["sub01"],
|
||||
env={"kind": "conda", "name": "junifer", "shell": shell},
|
||||
pre_run_cmds=pre_run,
|
||||
pre_run=pre_run,
|
||||
)
|
||||
assert shell in adapter.pre_run()
|
||||
assert expected_text in adapter.pre_run()
|
||||
assert shell in adapter.pre_run_cmds()
|
||||
assert expected_text in adapter.pre_run_cmds()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -111,10 +111,10 @@ def test_GnuParallelLocalAdapter_pre_collect(
|
|||
yaml_config_path=Path("."),
|
||||
elements=["sub01"],
|
||||
env={"kind": "venv", "name": "junifer", "shell": shell},
|
||||
pre_collect_cmds=pre_collect,
|
||||
pre_collect=pre_collect,
|
||||
)
|
||||
assert shell in adapter.pre_collect()
|
||||
assert expected_text in adapter.pre_collect()
|
||||
assert shell in adapter.pre_collect_cmds()
|
||||
assert expected_text in adapter.pre_collect_cmds()
|
||||
|
||||
|
||||
def test_GnuParallelLocalAdapter_run() -> None:
|
||||
|
|
@ -125,7 +125,7 @@ def test_GnuParallelLocalAdapter_run() -> None:
|
|||
yaml_config_path=Path("."),
|
||||
elements=["sub01"],
|
||||
)
|
||||
assert "run" in adapter.run()
|
||||
assert "run" in adapter.run_cmds()
|
||||
|
||||
|
||||
def test_GnuParallelLocalAdapter_collect() -> None:
|
||||
|
|
@ -136,7 +136,7 @@ def test_GnuParallelLocalAdapter_collect() -> None:
|
|||
yaml_config_path=Path("."),
|
||||
elements=["sub01"],
|
||||
)
|
||||
assert "collect" in adapter.collect()
|
||||
assert "collect" in adapter.collect_cmds()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
|
|||
|
|
@ -43,10 +43,10 @@ def test_HTCondorAdapter_pre_run(
|
|||
yaml_config_path=Path("."),
|
||||
elements=["sub01"],
|
||||
env={"kind": "conda", "name": "junifer", "shell": shell},
|
||||
pre_run_cmds=pre_run,
|
||||
pre_run=pre_run,
|
||||
)
|
||||
assert shell in adapter.pre_run()
|
||||
assert expected_text in adapter.pre_run()
|
||||
assert shell in adapter.pre_run_cmds()
|
||||
assert expected_text in adapter.pre_run_cmds()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -88,11 +88,11 @@ def test_HTCondorAdapter_pre_collect(
|
|||
yaml_config_path=Path("."),
|
||||
elements=["sub01"],
|
||||
env={"kind": "venv", "name": "junifer", "shell": shell},
|
||||
pre_collect_cmds=pre_collect,
|
||||
pre_collect=pre_collect,
|
||||
collect_task=collect,
|
||||
)
|
||||
assert shell in adapter.pre_collect()
|
||||
assert expected_text in adapter.pre_collect()
|
||||
assert shell in adapter.pre_collect_cmds()
|
||||
assert expected_text in adapter.pre_collect_cmds()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -122,8 +122,8 @@ def test_HTCondorAdapter_run_collect(
|
|||
elements=["sub01"],
|
||||
extra_preamble=extra_preamble,
|
||||
)
|
||||
assert expected_text in adapter.run()
|
||||
assert expected_text in adapter.collect()
|
||||
assert expected_text in adapter.run_cmds()
|
||||
assert expected_text in adapter.collect_cmds()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -165,7 +165,7 @@ def test_HTCondor_dag(
|
|||
elements=elements,
|
||||
collect_task=collect,
|
||||
)
|
||||
assert expected_text in adapter.dag()
|
||||
assert expected_text in adapter.dag_cmds()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
|
|||
Loading…
Reference in a new issue