juaml/junifer
Watch
3
0
Fork
You've already forked junifer
 0
Code Issues 14 Pull requests 2 Packages Activity Actions

[ENH]: Update QueueContextAdapter interface for backward compatibility #503

Merged
synchon merged 3 commits from fix/queue-context-interface into main 2026-07-21 10:41:38 +00:00
Conversation 1 Commits 3 Files changed 6 +59 -52
6 changed files with 59 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/changes/newsfragments/503.enh Normal file
View file

@ -0,0 +1 @@
Revert ``pre_run`` and ``pre_collect`` parameters for queue context adapters by `Synchon Mandal`_

30
junifer/api/queue_context/gnu_parallel_local_adapter.py
View file

@ -35,9 +35,9 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
The path to the YAML config file. The path to the YAML config file.
elements : ``Elements`` elements : ``Elements``
Element(s) to process. Will be used to index the DataGrabber. 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). 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). Extra shell commands to source before the collect (default None).
env : :class:`.QueueContextEnv` or None, optional env : :class:`.QueueContextEnv` or None, optional
The environment configuration. If None, will run without a The environment configuration. If None, will run without a
@ -63,8 +63,8 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
job_dir: Path job_dir: Path
yaml_config_path: Path yaml_config_path: Path
elements: Elements elements: Elements
pre_run_cmds: str | None = None pre_run: str | None = None
pre_collect_cmds: str | None = None pre_collect: str | None = None
env: QueueContextEnv | None = None env: QueueContextEnv | None = None
verbose: str = "info" verbose: str = "info"
verbose_datalad: str | None = None verbose_datalad: str | None = None
@ -105,7 +105,7 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
return "\n".join(elements_to_run) return "\n".join(elements_to_run)
def pre_run(self) -> str: def pre_run_cmds(self) -> str:
"""Return pre-run commands.""" """Return pre-run commands."""
fixed = ( fixed = (
f"#!/usr/bin/env {self.env['shell']}\n\n" 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" "# Force datalad to run in non-interactive mode\n"
"DATALAD_UI_INTERACTIVE=false\n" "DATALAD_UI_INTERACTIVE=false\n"
) )
var = self.pre_run_cmds or "" var = self.pre_run or ""
return fixed + "\n" + var return fixed + "\n" + var
def run(self) -> str: def run_cmds(self) -> str:
"""Return run commands.""" """Return run commands."""
verbose_args = f"--verbose {self.verbose}" verbose_args = f"--verbose {self.verbose}"
if self.verbose_datalad: if self.verbose_datalad:
@ -141,16 +141,16 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
f"--element" f"--element"
) )
def pre_collect(self) -> str: def pre_collect_cmds(self) -> str:
"""Return pre-collect commands.""" """Return pre-collect commands."""
fixed = ( fixed = (
f"#!/usr/bin/env {self.env['shell']}\n\n" f"#!/usr/bin/env {self.env['shell']}\n\n"
"# This script is auto-generated by junifer.\n" "# This script is auto-generated by junifer.\n"
) )
var = self.pre_collect_cmds or "" var = self.pre_collect or ""
return fixed + "\n" + var return fixed + "\n" + var
def collect(self) -> str: def collect_cmds(self) -> str:
"""Return collect commands.""" """Return collect commands."""
verbose_args = f"--verbose {self.verbose}" verbose_args = f"--verbose {self.verbose}"
if self.verbose_datalad: if self.verbose_datalad:
@ -196,14 +196,14 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
f"Writing {self._pre_run_path.name} to {self.job_dir.resolve()!s}" f"Writing {self._pre_run_path.name} to {self.job_dir.resolve()!s}"
) )
self._pre_run_path.touch() 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) make_executable(self._pre_run_path)
# Create run # Create run
logger.info( logger.info(
f"Writing {self._run_path.name} to {self.job_dir.resolve()!s}" f"Writing {self._run_path.name} to {self.job_dir.resolve()!s}"
) )
self._run_path.touch() 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) make_executable(self._run_path)
# Create pre collect # Create pre collect
logger.info( logger.info(
@ -211,14 +211,16 @@ class GnuParallelLocalAdapter(QueueContextAdapter):
f"{self.job_dir.resolve()!s}" f"{self.job_dir.resolve()!s}"
) )
self._pre_collect_path.touch() 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) make_executable(self._pre_collect_path)
# Create collect # Create collect
logger.info( logger.info(
f"Writing {self._collect_path.name} to {self.job_dir.resolve()!s}" f"Writing {self._collect_path.name} to {self.job_dir.resolve()!s}"
) )
self._collect_path.touch() 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) make_executable(self._collect_path)
# Submit if required # Submit if required
run_cmd = f"sh {self._run_path.resolve()!s}" run_cmd = f"sh {self._run_path.resolve()!s}"

36
junifer/api/queue_context/htcondor_adapter.py
View file

@ -52,9 +52,9 @@ class HTCondorAdapter(QueueContextAdapter):
The path to the YAML config file. The path to the YAML config file.
elements : ``Elements`` elements : ``Elements``
Element(s) to process. Will be used to index the DataGrabber. 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). 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). Extra shell commands to source before the collect (default None).
env : :class:`.QueueContextEnv` or None, optional env : :class:`.QueueContextEnv` or None, optional
The environment configuration. If None, will run without a The environment configuration. If None, will run without a
@ -91,8 +91,8 @@ class HTCondorAdapter(QueueContextAdapter):
job_dir: Path job_dir: Path
yaml_config_path: Path yaml_config_path: Path
elements: Elements elements: Elements
pre_run_cmds: str | None = None pre_run: str | None = None
pre_collect_cmds: str | None = None pre_collect: str | None = None
env: QueueContextEnv | None = None env: QueueContextEnv | None = None
verbose: str = "info" verbose: str = "info"
verbose_datalad: str | None = None verbose_datalad: str | None = None
@ -127,7 +127,7 @@ class HTCondorAdapter(QueueContextAdapter):
) )
self._dag_path = self.job_dir / f"{self.job_name}.dag" 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.""" """Return pre-run commands."""
fixed = ( fixed = (
f"#!/usr/bin/env {self.env['shell']}\n\n" 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" "# Force datalad to run in non-interactive mode\n"
"DATALAD_UI_INTERACTIVE=false\n" "DATALAD_UI_INTERACTIVE=false\n"
) )
var = self.pre_run_cmds or "" var = self.pre_run or ""
return fixed + "\n" + var return fixed + "\n" + var
def run(self) -> str: def run_cmds(self) -> str:
"""Return run commands.""" """Return run commands."""
verbose_args = f"--verbose {self.verbose} " verbose_args = f"--verbose {self.verbose} "
if self.verbose_datalad is not None: if self.verbose_datalad is not None:
@ -176,19 +176,19 @@ class HTCondorAdapter(QueueContextAdapter):
var = self.extra_preamble or "" var = self.extra_preamble or ""
return fixed + "\n" + var + "\n" + "queue" return fixed + "\n" + var + "\n" + "queue"
def pre_collect(self) -> str: def pre_collect_cmds(self) -> str:
"""Return pre-collect commands.""" """Return pre-collect commands."""
fixed = ( fixed = (
f"#!/usr/bin/env {self.env['shell']}\n\n" f"#!/usr/bin/env {self.env['shell']}\n\n"
"# This script is auto-generated by junifer.\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" # Add commands if collect="yes"
if self.collect_task == "yes": if self.collect_task == "yes":
var += 'if [ "${1}" == "4" ]; then\n exit 1\nfi\n' var += 'if [ "${1}" == "4" ]; then\n exit 1\nfi\n'
return fixed + "\n" + var return fixed + "\n" + var
def collect(self) -> str: def collect_cmds(self) -> str:
"""Return collect commands.""" """Return collect commands."""
verbose_args = f"--verbose {self.verbose} " verbose_args = f"--verbose {self.verbose} "
if self.verbose_datalad is not None: if self.verbose_datalad is not None:
@ -222,7 +222,7 @@ class HTCondorAdapter(QueueContextAdapter):
var = self.extra_preamble or "" var = self.extra_preamble or ""
return fixed + "\n" + var + "\n" + "queue" return fixed + "\n" + var + "\n" + "queue"
def dag(self) -> str: def dag_cmds(self) -> str:
"""Return HTCondor DAG commands.""" """Return HTCondor DAG commands."""
fixed = "" fixed = ""
for idx, element in enumerate(self.elements): 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}" f"Writing {self._pre_run_path.name} to {self.job_dir.resolve()!s}"
) )
self._pre_run_path.touch() 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) make_executable(self._pre_run_path)
# Create run # Create run
logger.debug( logger.debug(
@ -285,14 +285,16 @@ class HTCondorAdapter(QueueContextAdapter):
f"{self.job_dir.resolve()!s}" f"{self.job_dir.resolve()!s}"
) )
self._submit_run_path.touch() 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 # Create pre collect
logger.info( logger.info(
f"Writing {self._pre_collect_path.name} to " f"Writing {self._pre_collect_path.name} to "
f"{self.job_dir.resolve()!s}" f"{self.job_dir.resolve()!s}"
) )
self._pre_collect_path.touch() 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) make_executable(self._pre_collect_path)
# Create collect # Create collect
logger.debug( logger.debug(
@ -300,13 +302,15 @@ class HTCondorAdapter(QueueContextAdapter):
f"{self.job_dir.resolve()!s}" f"{self.job_dir.resolve()!s}"
) )
self._submit_collect_path.touch() 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 # Create DAG
logger.debug( logger.debug(
f"Writing {self._dag_path.name} to {self.job_dir.resolve()!s}" f"Writing {self._dag_path.name} to {self.job_dir.resolve()!s}"
) )
self._dag_path.touch() 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 # Submit if required
condor_submit_dag_cmd = [ condor_submit_dag_cmd = [
"condor_submit_dag", "condor_submit_dag",

10
junifer/api/queue_context/queue_context_adapter.py
View file

@ -64,11 +64,11 @@ class QueueContextAdapter(BaseModel, ABC):
model_config = ConfigDict( model_config = ConfigDict(
use_enum_values=True, use_enum_values=True,
extra="allow", extra="forbid",
) )
@abstractmethod @abstractmethod
def pre_run(self) -> str: def pre_run_cmds(self) -> str:
"""Return pre-run commands.""" """Return pre-run commands."""
raise_error( raise_error(
msg="Concrete classes need to implement pre_run()", msg="Concrete classes need to implement pre_run()",
@ -76,7 +76,7 @@ class QueueContextAdapter(BaseModel, ABC):
) )
@abstractmethod @abstractmethod
def run(self) -> str: def run_cmds(self) -> str:
"""Return run commands.""" """Return run commands."""
raise_error( raise_error(
msg="Concrete classes need to implement run()", msg="Concrete classes need to implement run()",
@ -84,7 +84,7 @@ class QueueContextAdapter(BaseModel, ABC):
) )
@abstractmethod @abstractmethod
def pre_collect(self) -> str: def pre_collect_cmds(self) -> str:
"""Return pre-collect commands.""" """Return pre-collect commands."""
raise_error( raise_error(
msg="Concrete classes need to implement pre_collect()", msg="Concrete classes need to implement pre_collect()",
@ -92,7 +92,7 @@ class QueueContextAdapter(BaseModel, ABC):
) )
@abstractmethod @abstractmethod
def collect(self) -> str: def collect_cmds(self) -> str:
"""Return collect commands.""" """Return collect commands."""
raise_error( raise_error(
msg="Concrete classes need to implement collect()", msg="Concrete classes need to implement collect()",

16
junifer/api/queue_context/tests/test_gnu_parallel_local_adapter.py
View file

@ -73,10 +73,10 @@ def test_GnuParallelLocalAdapter_pre_run(
yaml_config_path=Path("."), yaml_config_path=Path("."),
elements=["sub01"], elements=["sub01"],
env={"kind": "conda", "name": "junifer", "shell": shell}, env={"kind": "conda", "name": "junifer", "shell": shell},
pre_run_cmds=pre_run, pre_run=pre_run,
) )
assert shell in adapter.pre_run() assert shell in adapter.pre_run_cmds()
assert expected_text in adapter.pre_run() assert expected_text in adapter.pre_run_cmds()
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -111,10 +111,10 @@ def test_GnuParallelLocalAdapter_pre_collect(
yaml_config_path=Path("."), yaml_config_path=Path("."),
elements=["sub01"], elements=["sub01"],
env={"kind": "venv", "name": "junifer", "shell": shell}, env={"kind": "venv", "name": "junifer", "shell": shell},
pre_collect_cmds=pre_collect, pre_collect=pre_collect,
) )
assert shell in adapter.pre_collect() assert shell in adapter.pre_collect_cmds()
assert expected_text in adapter.pre_collect() assert expected_text in adapter.pre_collect_cmds()
def test_GnuParallelLocalAdapter_run() -> None: def test_GnuParallelLocalAdapter_run() -> None:
@ -125,7 +125,7 @@ def test_GnuParallelLocalAdapter_run() -> None:
yaml_config_path=Path("."), yaml_config_path=Path("."),
elements=["sub01"], elements=["sub01"],
) )
assert "run" in adapter.run() assert "run" in adapter.run_cmds()
def test_GnuParallelLocalAdapter_collect() -> None: def test_GnuParallelLocalAdapter_collect() -> None:
@ -136,7 +136,7 @@ def test_GnuParallelLocalAdapter_collect() -> None:
yaml_config_path=Path("."), yaml_config_path=Path("."),
elements=["sub01"], elements=["sub01"],
) )
assert "collect" in adapter.collect() assert "collect" in adapter.collect_cmds()
@pytest.mark.parametrize( @pytest.mark.parametrize(

18
junifer/api/queue_context/tests/test_htcondor_adapter.py
View file

@ -43,10 +43,10 @@ def test_HTCondorAdapter_pre_run(
yaml_config_path=Path("."), yaml_config_path=Path("."),
elements=["sub01"], elements=["sub01"],
env={"kind": "conda", "name": "junifer", "shell": shell}, env={"kind": "conda", "name": "junifer", "shell": shell},
pre_run_cmds=pre_run, pre_run=pre_run,
) )
assert shell in adapter.pre_run() assert shell in adapter.pre_run_cmds()
assert expected_text in adapter.pre_run() assert expected_text in adapter.pre_run_cmds()
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -88,11 +88,11 @@ def test_HTCondorAdapter_pre_collect(
yaml_config_path=Path("."), yaml_config_path=Path("."),
elements=["sub01"], elements=["sub01"],
env={"kind": "venv", "name": "junifer", "shell": shell}, env={"kind": "venv", "name": "junifer", "shell": shell},
pre_collect_cmds=pre_collect, pre_collect=pre_collect,
collect_task=collect, collect_task=collect,
) )
assert shell in adapter.pre_collect() assert shell in adapter.pre_collect_cmds()
assert expected_text in adapter.pre_collect() assert expected_text in adapter.pre_collect_cmds()
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -122,8 +122,8 @@ def test_HTCondorAdapter_run_collect(
elements=["sub01"], elements=["sub01"],
extra_preamble=extra_preamble, extra_preamble=extra_preamble,
) )
assert expected_text in adapter.run() assert expected_text in adapter.run_cmds()
assert expected_text in adapter.collect() assert expected_text in adapter.collect_cmds()
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -165,7 +165,7 @@ def test_HTCondor_dag(
elements=elements, elements=elements,
collect_task=collect, collect_task=collect,
) )
assert expected_text in adapter.dag() assert expected_text in adapter.dag_cmds()
@pytest.mark.parametrize( @pytest.mark.parametrize(