From cab9af240fe11e5b3974fd4e341d5fae84dc58bd Mon Sep 17 00:00:00 2001 From: Satrajit Ghosh Date: Sat, 21 Feb 2026 09:33:14 -0500 Subject: [PATCH 1/8] enh: reduce per-task overhead (~26x speedup for simple tasks) Five targeted optimisations identified via cProfile on the audio-file benchmark from issue #850: 1. **Etelemetry sentinel** (`job.py`, `submitter.py`): Replace the `None` initial value of `Job._etelemetry_version_data` with a distinct `_ETELEMETRY_UNCHECKED` sentinel. Previously a failed network check returned `None`, keeping the `is None` guard True forever and re-issuing the HTTP request on every task (~93 ms each). 2. **Function bytes cache** (`hash.py`): Add a module-level `_function_bytes_cache` keyed by `(module, qualname, mtime_ns)`. `inspect.getsource()` + `ast.parse()` now runs at most once per function per session; subsequent calls cost only a single `os.stat()`. 3. **Skip hash-change check for non-FileSet tasks** (`job.py`): Call `TypeParser.contains_type(FileSet, ...)` on each input field; if none match, skip the expensive full hash recomputation in `_check_for_hash_changes()`. Scalar/pure-Python values cannot mutate under Pydra. 4. **In-memory result cache** (`job.py`): Store the completed `Result` on `self._cached_result` at the end of `run()` so same-process callers (e.g. `Submitter.__call__` with DebugWorker) do not need to deserialise it back from disk. The field is excluded from `__getstate__` so subprocess/Slurm workers still use the disk path. 5. **Once-per-location PersistentCache.clean_up()** (`hash.py`): Track which cache locations have already been scanned in a class-level set (`_session_cleanups_done`). The O(n) `iterdir()` + `stat()` loop no longer runs after every task. `path.unlink(missing_ok=True)` makes concurrent cleanup by multiple Slurm nodes on shared NFS safe. Co-Authored-By: Claude Sonnet 4.6 --- CLAUDE.md | 196 ++++++++++++++++++++++++++++++++++++++ pydra/engine/job.py | 43 ++++++++- pydra/engine/submitter.py | 2 +- pydra/utils/hash.py | 118 ++++++++++++++++------- 4 files changed, 321 insertions(+), 38 deletions(-) create mode 100644 CLAUDE.md diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 000000000..d5155f22e --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,196 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Overview + +Pydra is a dataflow engine for constructing and executing directed acyclic graphs (DAGs) of tasks. It is the core for Nipype 2.0. Requires Python 3.11+. Uses `attrs` extensively for dataclass-like definitions and `hatchling` + `hatch-vcs` as the build system (version derived from git tags). + +## Commands + +### Install + +```bash +pip install -e ".[dev]" # full dev install (includes test + lint deps) +pip install -e ".[test]" # test deps only +pip install -e ".[doc]" # doc deps only +``` + +### Testing + +```bash +pytest pydra # full test suite (parallel, with coverage) +pytest pydra/engine/tests/test_job.py # single test file +pytest pydra/engine/tests/test_job.py::test_my_func # single test +pytest pydra --only-worker=debug # single-process worker (good for debugging) +pytest pydra --only-worker=cf # ConcurrentFutures worker +pytest pydra --only-worker=slurm # SLURM (requires sbatch) +pytest pydra --with-dask # Dask worker +``` + +Set `_PYTEST_RAISE=1` for IDE breakpoint-friendly exception propagation. + +Tests include doctests (`--doctest-modules` is on). `xfail_strict = true` means unexpected passes fail. + +### Linting / Formatting + +```bash +tox -e style # check style (ruff) +tox -e style-fix # auto-fix style +tox -e spellcheck # codespell check +pre-commit run --all-files # black + flake8 + codespell + nbstripout +black pydra +flake8 # max-line-length=105, ignores E203/W503/F541 +``` + +### Tox Environments + +```bash +tox -e py311-latest # Python 3.11, latest deps +tox -e py313-pre # Python 3.13, pre-release deps +tox -e py311-min # Python 3.11, minimum pinned deps +``` + +### Docs + +```bash +make -C docs html +``` + +### CLI + +```bash +pydracli crash # display crash file +pydracli crash --rerun # rerun crashed job +``` + +## Architecture + +### Layer Overview + +``` +compose/ Task definition (decorators + task types) +engine/ Execution engine (graph, jobs, state, submitter) +workers/ Execution backends (cf, debug, slurm, sge, + plugins) +environments/ Execution environments (native, docker, singularity, lmod) +utils/ Hashing, typing helpers, plugin discovery, profiling +tasks/ Built-in reusable tasks +scripts/ CLI entry points +``` + +### Task Definition (`compose/`) + +Three task flavors, each defined by a decorator: + +**Python tasks** — wrap a Python function: +```python +@python.define +def Add(a: int, b: int) -> int: + return a + b +``` + +**Shell tasks** — wrap a CLI command: +```python +@shell.define +class BET(shell.Task["BET.Outputs"]): + executable = "bet" + input_image: File = shell.arg(argstr="{input_image}", position=1) +``` + +**Workflow tasks** — compose other tasks into a DAG: +```python +@workflow.define +def MyWorkflow(x: int) -> int: + node_a = workflow.add(Add(a=x, b=1)) + return node_a.out +``` + +The decorator machinery is in `compose/base/builder.py` (`build_task_class()`). It converts `Arg`/`Out` field specs into `attrs` fields and dynamically creates a `Task` class + a paired `Outputs` class. + +Key base types: `compose.base.Field`, `Arg`, `Out`, `Task[OutputType]`, `Outputs`. + +### Execution Engine (`engine/`) + +**Workflow construction** (`engine/workflow.py`): `Workflow.construct(task)` runs the workflow definition function to discover nodes and wire the `DiGraph`. Constructed workflows are cached by content hash. + +**Node** (`engine/node.py`): Wraps a `Task` inside a workflow. Holds the task, its `State`, optional `Environment`, and `TaskHooks`. Exposes `lzout` — a lazy output proxy for wiring downstream nodes. + +**LazyField** (`engine/lazy.py`): Promises between nodes. `node.lzout.x` returns a `LazyOutField`. Assigning it to another node's input creates a dataflow edge. + +**State / Splitter / Combiner** (`engine/state.py`): Implements map-reduce semantics. A node can be split over an iterable input (producing parallel jobs) and combined (reducing results). Splitters can be scalar (zip) or outer (cartesian product), expressed in RPN. Each concrete state index corresponds to one `Job`. + +**Job** (`engine/job.py`): The concrete unit of work submitted to a worker. Holds the fully-resolved `Task`, a `cache_dir` (from content hash of task inputs + definition), and uses `filelock.SoftFileLock` for safe parallel execution. + +**Submitter** (`engine/submitter.py`): The async dispatch loop. Constructs the `DiGraph` of `NodeExecution` objects, drives an asyncio event loop to submit ready jobs to the configured `Worker`, handles caching (skips jobs with a valid cached result), and manages concurrency. + +```python +with Submitter(worker="cf", cache_root="/tmp/cache") as sub: + result = sub(my_task) +``` + +### Workers (`workers/`) + +| Class | Module | Description | +|---|---|---| +| `ConcurrentFuturesWorker` | `cf.py` | `ProcessPoolExecutor`-based (default) | +| `DebugWorker` | `debug.py` | Single-process, synchronous | +| `SlurmWorker` | `slurm.py` | Submits via `sbatch`, polls with `sacct` | +| `SGEWorker` | `sge.py` | SGE qsub | + +Workers are discovered via a plugin system (`get_plugin_classes` in `utils/general.py`). External workers (`pydra-workers-psij`, `pydra-workers-dask`) are installable as separate packages. + +### Environments (`environments/`) + +Control *how* shell tasks execute: `native.py` (bare OS), `docker.py`, `singularity.py`, `lmod.py` (load environment modules before executing). + +### Caching (`utils/hash.py`, `engine/job.py`) + +Cache keys are content hashes of the `Task` (all inputs + task definition). Results are stored as cloudpickled files under `~/.cache/pydra//run-cache/` (via `platformdirs`). Lock files prevent race conditions. + +### Provenance Tracking (`engine/audit.py`) + +Optional JSON-LD provenance tracking controlled via `AuditFlag` bits. Messengers: `PrintMessenger`, `FileMessenger`, `RemoteRESTMessenger`. Schema at `schema/context.jsonld`. + +### Data Flow Summary + +``` +User code + │ + ├─ @python.define / @shell.define / @workflow.define + │ └─> compose/base/builder.py: build_task_class() + │ creates Task(attrs) + Outputs(attrs) + │ + ├─ Submitter(worker="cf", cache_root=...) + │ ├─ Workflow.construct(task) → DiGraph of Nodes (engine/graph.py) + │ ├─ State resolution → list of state indices (engine/state.py) + │ ├─ per state-index: Job(task, cache_dir) (engine/job.py) + │ │ └─ if not cached → Worker.run(job) (workers/) + │ │ └─ Environment.execute(job) (environments/) + │ └─ Result(outputs, runtime, cache_dir) (engine/result.py) + │ + └─ LazyField wiring between Nodes (engine/lazy.py) +``` + +## Key Files + +| File | Purpose | +|---|---| +| `pyproject.toml` | Build, deps, pytest config, coverage config | +| `tox.ini` | tox envs: test, style, style-fix, spellcheck, build, publish | +| `.flake8` | Flake8: max-line-length=105 | +| `pydra/conftest.py` | `worker`/`any_worker` fixtures; `--only-worker`, `--with-dask` flags | +| `pydra/compose/base/builder.py` | Decorator machinery (`build_task_class`) | +| `pydra/compose/base/field.py` | `Field`, `Arg`, `Out`, `NO_DEFAULT`, `Requirement` | +| `pydra/compose/base/task.py` | `Task` and `Outputs` base classes | +| `pydra/compose/python.py` | `@python.define` | +| `pydra/compose/shell/task.py` | Shell `Task`, CLI construction | +| `pydra/compose/workflow.py` | `@workflow.define`, `workflow.add`, `workflow.this` | +| `pydra/engine/submitter.py` | Async dispatch loop | +| `pydra/engine/job.py` | Single unit of work, caching, locking | +| `pydra/engine/state.py` | Splitter/combiner map-reduce | +| `pydra/engine/lazy.py` | `LazyField` — dataflow wiring | +| `pydra/engine/workflow.py` | DAG construction and caching | +| `pydra/engine/node.py` | `Node` — task wrapper in workflow graph | +| `pydra/utils/hash.py` | Content hashing for cache keys | +| `pydra/utils/general.py` | Plugin discovery, cache root, platform utils | +| `pydra/utils/typing.py` | `StateArray`, `TypeParser`, type helpers | diff --git a/pydra/engine/job.py b/pydra/engine/job.py index e27891f1f..dd1c8b6f7 100644 --- a/pydra/engine/job.py +++ b/pydra/engine/job.py @@ -59,7 +59,11 @@ class Job(ty.Generic[TaskType]): """ _api_version: str = "0.0.1" # Should generally not be touched by subclasses - _etelemetry_version_data = None # class variable to store etelemetry information + # Sentinel meaning "check not yet performed". Distinct from None so that a + # failed/empty etelemetry response (which returns None) does not cause the + # network check to be repeated on every subsequent task invocation. + _ETELEMETRY_UNCHECKED = object() + _etelemetry_version_data = _ETELEMETRY_UNCHECKED # class variable _version: str # Version of tool being wrapped _task_version: ty.Optional[str] = None # Job writers encouraged to define and increment when implementation changes sufficiently @@ -141,6 +145,10 @@ def __init__( self.hooks = hooks if hooks is not None else TaskHooks() self._errored = False self._lzout = None + # In-memory result cache: avoids reading back from disk the result that + # was just written by run(). Not included in __getstate__ so it is + # never pickled (subprocess workers will still use the disk path). + self._cached_result = None # Save the submitter attributes needed to run the job later self.audit = submitter.audit @@ -179,11 +187,18 @@ def __str__(self): def __getstate__(self): state = self.__dict__.copy() state["task"] = cp.dumps(state["task"]) + # Never serialise the in-memory result cache: subprocess workers must + # go through the normal disk-based load_result() path. + state.pop("_cached_result", None) return state def __setstate__(self, state): state["task"] = cp.loads(state["task"]) self.__dict__.update(state) + # _cached_result is excluded from __getstate__; ensure it always exists + # so that result() works correctly on deserialized jobs. + if "_cached_result" not in self.__dict__: + self._cached_result = None @property def errored(self): @@ -349,6 +364,10 @@ def run(self, rerun: bool = False): # Check for any changes to the input hashes that have occurred during the execution # of the job self._check_for_hash_changes() + # Cache the completed result in memory so that callers who immediately + # call job.result() (e.g. Submitter.__call__) do not need to deserialise + # it back from disk. + self._cached_result = result return result async def run_async(self, rerun: bool = False) -> Result: @@ -487,6 +506,11 @@ def result(self, return_inputs=False): task=self.task, ) + # Fast path: return the in-memory result cached by run() to avoid + # deserialising from disk for callers in the same process. + if self._cached_result is not None and not return_inputs: + return self._cached_result + checksum = self.checksum result = load_result(checksum, self.all_caches) if result and result.errored: @@ -504,6 +528,23 @@ def result(self, return_inputs=False): return result def _check_for_hash_changes(self): + from pydra.utils.typing import TypeParser + + # For tasks whose input fields contain no FileSet types, hashes cannot + # change during execution (scalar/pure-Python values are immutable from + # Pydra's perspective). Skip the expensive full recomputation in that + # common case. + if not any( + TypeParser.contains_type(FileSet, f.type) for f in get_fields(self.task) + ): + logger.debug( + "Input values and hashes for '%s' %s node:\n%s\n%s", + self.name, + type(self).__name__, + self.task, + self.task._hashes, + ) + return hash_changes = self.task._hash_changes() details = "" for changed in hash_changes: diff --git a/pydra/engine/submitter.py b/pydra/engine/submitter.py index 28e3d566e..f97b22002 100644 --- a/pydra/engine/submitter.py +++ b/pydra/engine/submitter.py @@ -115,7 +115,7 @@ def __init__( from pydra.utils.etelemetry import check_latest_version - if Job._etelemetry_version_data is None: + if Job._etelemetry_version_data is Job._ETELEMETRY_UNCHECKED: Job._etelemetry_version_data = check_latest_version() self.audit = Audit( diff --git a/pydra/utils/hash.py b/pydra/utils/hash.py index 326e9c679..5c5193d11 100644 --- a/pydra/utils/hash.py +++ b/pydra/utils/hash.py @@ -32,6 +32,11 @@ FUNCTION_SRC_CHUNK_LEN_DEFAULT = 8192 +# Module-level cache for function source bytes to avoid repeated disk I/O and +# AST parsing. Key: (module, qualname, source_file_mtime_ns); value: bytes of +# the function's inner content (between "function:(" and ")"). +_function_bytes_cache: dict[tuple, bytes] = {} + try: from typing import Protocol except ImportError: @@ -102,6 +107,12 @@ class PersistentCache: cleanup_period: int = attrs.field() _hashes: ty.Dict[CacheKey, Hash] = attrs.field(factory=dict) + # Class-level guard so clean_up() runs at most once per Python session per + # cache location. Iterating and stat()-ing every entry in the cache + # directory is expensive when the cache is large; doing it after every task + # invocation causes significant overhead for batch workloads. + _session_cleanups_done: ty.ClassVar[Set[Path]] = set() + # Set the location of the persistent hash cache LOCATION_ENV_VAR = "PYDRA_HASH_CACHE" CLEANUP_ENV_VAR = "PYDRA_HASH_CACHE_CLEANUP_PERIOD" @@ -156,14 +167,22 @@ def get_or_calculate_hash(self, key: CacheKey, calculate_hash: ty.Callable) -> H return Hash(hsh) def clean_up(self): - """Cleans up old hash caches that haven't been accessed in the last 30 days""" + """Cleans up old hash caches that haven't been accessed in the last 30 days. + + Cleanup for a given cache location runs at most once per Python session + to avoid the O(n) directory scan becoming a per-task overhead for large + persistent caches. + """ + if self.location in PersistentCache._session_cleanups_done: + return + PersistentCache._session_cleanups_done.add(self.location) now = datetime.now() for path in self.location.iterdir(): if path.name.endswith(".lock"): continue days = (now - datetime.fromtimestamp(path.lstat().st_atime)).days if days > self.cleanup_period: - path.unlink() + path.unlink(missing_ok=True) @classmethod def from_path( @@ -615,6 +634,45 @@ def bytes_repr_code(obj: types.CodeType, cache: Cache) -> Iterator[bytes]: yield b")" +def _parse_function_source_bytes(src: str) -> bytes: + """Parse function source code into a stable bytes representation. + + Strips type annotations (which may reference objects that are not stable + across runs) and returns AST-dump bytes, falling back to raw source bytes + on a SyntaxError. + """ + + def dump_ast(node: ast.AST) -> bytes: + return ast.dump(node, annotate_fields=False, include_attributes=False).encode() + + def strip_annotations(node: ast.AST) -> None: + if hasattr(node, "args"): + for arg in node.args.args: + arg.annotation = None + for arg in node.args.kwonlyargs: + arg.annotation = None + if node.args.vararg: + node.args.vararg.annotation = None + if node.args.kwarg: + node.args.kwarg.annotation = None + + indent = re.match(r"(\s*)", src).group(1) + if indent: + src = re.sub(f"^{indent}", "", src, flags=re.MULTILINE) + parts: list[bytes] = [] + try: + func_ast = ast.parse(src).body[0] + strip_annotations(func_ast) + if hasattr(func_ast, "args"): + parts.append(dump_ast(func_ast.args)) + if hasattr(func_ast, "body"): + for stmt in func_ast.body: + parts.append(dump_ast(stmt)) + except SyntaxError: + parts.append(src.encode()) + return b"".join(parts) + + @register_serializer def bytes_repr_function(obj: types.FunctionType, cache: Cache) -> Iterator[bytes]: """Serialize a function, attempting to use the AST of the source code if available @@ -623,43 +681,31 @@ def bytes_repr_function(obj: types.FunctionType, cache: Cache) -> Iterator[bytes if in_stdlib(obj): yield f"{obj.__module__}.{obj.__name__}".encode() else: + # Build a cache key from (module, qualname, source-file mtime). + # os.stat() is a single cheap syscall; if successful we can avoid + # both inspect.getsource() and ast.parse() on repeated calls. + cache_key: tuple | None = None try: - src = inspect.getsource(obj) - except OSError: - # Fallback to using the bytes representation of the code object - yield from bytes_repr(obj.__code__, cache) - else: + source_file = inspect.getfile(obj) + mtime_ns = os.stat(source_file).st_mtime_ns + cache_key = (obj.__module__, obj.__qualname__, mtime_ns) + except (OSError, TypeError): + pass - def dump_ast(node: ast.AST) -> bytes: - return ast.dump( - node, annotate_fields=False, include_attributes=False - ).encode() - - def strip_annotations(node: ast.AST): - """Remove annotations from function arguments.""" - if hasattr(node, "args"): - for arg in node.args.args: - arg.annotation = None - for arg in node.args.kwonlyargs: - arg.annotation = None - if node.args.vararg: - node.args.vararg.annotation = None - if node.args.kwarg: - node.args.kwarg.annotation = None - - indent = re.match(r"(\s*)", src).group(1) - if indent: - src = re.sub(f"^{indent}", "", src, flags=re.MULTILINE) + if cache_key is not None and cache_key in _function_bytes_cache: + yield _function_bytes_cache[cache_key] + else: try: - func_ast = ast.parse(src).body[0] - strip_annotations(func_ast) - if hasattr(func_ast, "args"): - yield dump_ast(func_ast.args) - if hasattr(func_ast, "body"): - for stmt in func_ast.body: - yield dump_ast(stmt) - except SyntaxError: - yield src.encode() + src = inspect.getsource(obj) + except OSError: + # Fallback to using the bytes representation of the code object. + # This path is not cached because it depends on the cache arg. + yield from bytes_repr(obj.__code__, cache) + else: + result = _parse_function_source_bytes(src) + if cache_key is not None: + _function_bytes_cache[cache_key] = result + yield result yield b")" From 81dca6429e7080334102facb36f38522db845989 Mon Sep 17 00:00:00 2001 From: Satrajit Ghosh Date: Thu, 5 Mar 2026 18:48:40 -0500 Subject: [PATCH 2/8] fix: detect unstable hashes for custom __bytes_repr__ types The early-return optimization in _check_for_hash_changes() skipped the hash recomputation for tasks with no FileSet inputs, assuming non-FileSet values are always stable. This broke detection of unstable hashes for custom types implementing __bytes_repr__ (HasBytesRepr protocol). Fix by also checking whether any input value implements HasBytesRepr; if so, fall through to the full hash-change check. Co-Authored-By: Claude Sonnet 4.6 --- pydra/engine/job.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/pydra/engine/job.py b/pydra/engine/job.py index dd1c8b6f7..59a9f86df 100644 --- a/pydra/engine/job.py +++ b/pydra/engine/job.py @@ -529,13 +529,17 @@ def result(self, return_inputs=False): def _check_for_hash_changes(self): from pydra.utils.typing import TypeParser + from pydra.utils.hash import HasBytesRepr - # For tasks whose input fields contain no FileSet types, hashes cannot - # change during execution (scalar/pure-Python values are immutable from - # Pydra's perspective). Skip the expensive full recomputation in that - # common case. + # For tasks whose input fields contain no FileSet types and no values + # with custom __bytes_repr__ methods, hashes cannot change during + # execution (scalar/pure-Python values are immutable from Pydra's + # perspective). Skip the expensive full recomputation in that common case. if not any( TypeParser.contains_type(FileSet, f.type) for f in get_fields(self.task) + ) and not any( + isinstance(getattr(self.task, f.name), HasBytesRepr) + for f in get_fields(self.task) ): logger.debug( "Input values and hashes for '%s' %s node:\n%s\n%s", From f95951e289fd31ff3f35718b09911766eeeb38a3 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Fri, 6 Mar 2026 12:29:25 +0000 Subject: [PATCH 3/8] refactor: encapsulate etelemetry sentinel access in Job.check_etelemetry() Move the private _etelemetry_version_data / _ETELEMETRY_UNCHECKED access from submitter.py into a new Job.check_etelemetry() classmethod, so the sentinel pattern stays an implementation detail of Job and Submitter only calls a clean public API. Co-authored-by: Satrajit Ghosh Co-Authored-By: Claude Sonnet 4.6 --- pydra/engine/job.py | 8 ++++++++ pydra/engine/submitter.py | 5 +---- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/pydra/engine/job.py b/pydra/engine/job.py index 59a9f86df..716be93db 100644 --- a/pydra/engine/job.py +++ b/pydra/engine/job.py @@ -200,6 +200,14 @@ def __setstate__(self, state): if "_cached_result" not in self.__dict__: self._cached_result = None + @classmethod + def check_etelemetry(cls) -> None: + """Run the etelemetry version check at most once per session.""" + from pydra.utils.etelemetry import check_latest_version + + if cls._etelemetry_version_data is cls._ETELEMETRY_UNCHECKED: + cls._etelemetry_version_data = check_latest_version() + @property def errored(self): """Check if the job has raised an error""" diff --git a/pydra/engine/submitter.py b/pydra/engine/submitter.py index f97b22002..7e4c34cf9 100644 --- a/pydra/engine/submitter.py +++ b/pydra/engine/submitter.py @@ -113,10 +113,7 @@ def __init__( if worker is None: worker = "debug" - from pydra.utils.etelemetry import check_latest_version - - if Job._etelemetry_version_data is Job._ETELEMETRY_UNCHECKED: - Job._etelemetry_version_data = check_latest_version() + Job.check_etelemetry() self.audit = Audit( audit_flags=audit_flags, From af6e003f9fb3ff4c546d380f89353dff15188e53 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Fri, 6 Mar 2026 13:23:10 +0000 Subject: [PATCH 4/8] refactor: move etelemetry import to top of job.py Co-authored-by: Satrajit Ghosh Co-Authored-By: Claude Sonnet 4.6 --- pydra/engine/job.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/pydra/engine/job.py b/pydra/engine/job.py index 716be93db..9b64f9ca3 100644 --- a/pydra/engine/job.py +++ b/pydra/engine/job.py @@ -37,6 +37,7 @@ from pydra.compose.shell.templating import template_update from pydra.utils.messenger import AuditFlag from pydra.environments.base import Environment +from pydra.utils.etelemetry import check_latest_version logger = logging.getLogger("pydra") @@ -203,8 +204,6 @@ def __setstate__(self, state): @classmethod def check_etelemetry(cls) -> None: """Run the etelemetry version check at most once per session.""" - from pydra.utils.etelemetry import check_latest_version - if cls._etelemetry_version_data is cls._ETELEMETRY_UNCHECKED: cls._etelemetry_version_data = check_latest_version() From 4859b706f3bb5171bcb25eb13beea66600082500 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Fri, 6 Mar 2026 13:28:05 +0000 Subject: [PATCH 5/8] refactor: simplify etelemetry check to boolean flag Replace the sentinel object pattern with a plain boolean class variable, as the version data returned by check_latest_version() is never used by Pydra itself. Co-authored-by: Satrajit Ghosh --- pydra/engine/job.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/pydra/engine/job.py b/pydra/engine/job.py index 9b64f9ca3..2f04e2aa5 100644 --- a/pydra/engine/job.py +++ b/pydra/engine/job.py @@ -60,11 +60,7 @@ class Job(ty.Generic[TaskType]): """ _api_version: str = "0.0.1" # Should generally not be touched by subclasses - # Sentinel meaning "check not yet performed". Distinct from None so that a - # failed/empty etelemetry response (which returns None) does not cause the - # network check to be repeated on every subsequent task invocation. - _ETELEMETRY_UNCHECKED = object() - _etelemetry_version_data = _ETELEMETRY_UNCHECKED # class variable + _etelemetry_checked: ty.ClassVar[bool] = False _version: str # Version of tool being wrapped _task_version: ty.Optional[str] = None # Job writers encouraged to define and increment when implementation changes sufficiently @@ -204,8 +200,9 @@ def __setstate__(self, state): @classmethod def check_etelemetry(cls) -> None: """Run the etelemetry version check at most once per session.""" - if cls._etelemetry_version_data is cls._ETELEMETRY_UNCHECKED: - cls._etelemetry_version_data = check_latest_version() + if not cls._etelemetry_checked: + cls._etelemetry_checked = True + check_latest_version() @property def errored(self): From 590613103d0bae1d88a92c69eeb01b51b6545663 Mon Sep 17 00:00:00 2001 From: Satrajit Ghosh Date: Fri, 6 Mar 2026 10:49:58 -0500 Subject: [PATCH 6/8] Apply suggestions from code review Co-authored-by: Chris Markiewicz --- pydra/engine/job.py | 5 ++--- pydra/utils/hash.py | 2 +- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/pydra/engine/job.py b/pydra/engine/job.py index 2f04e2aa5..8bec91498 100644 --- a/pydra/engine/job.py +++ b/pydra/engine/job.py @@ -540,9 +540,8 @@ def _check_for_hash_changes(self): # execution (scalar/pure-Python values are immutable from Pydra's # perspective). Skip the expensive full recomputation in that common case. if not any( - TypeParser.contains_type(FileSet, f.type) for f in get_fields(self.task) - ) and not any( - isinstance(getattr(self.task, f.name), HasBytesRepr) + TypeParser.contains_type(FileSet, f.type) or + hasattr(getattr(self.task, f.name), '__bytes_repr__') for f in get_fields(self.task) ): logger.debug( diff --git a/pydra/utils/hash.py b/pydra/utils/hash.py index 5c5193d11..33211739c 100644 --- a/pydra/utils/hash.py +++ b/pydra/utils/hash.py @@ -684,7 +684,7 @@ def bytes_repr_function(obj: types.FunctionType, cache: Cache) -> Iterator[bytes # Build a cache key from (module, qualname, source-file mtime). # os.stat() is a single cheap syscall; if successful we can avoid # both inspect.getsource() and ast.parse() on repeated calls. - cache_key: tuple | None = None + cache_key: tuple[str, str, int] | None = None try: source_file = inspect.getfile(obj) mtime_ns = os.stat(source_file).st_mtime_ns From 3dcca84f8eeb576c9c30b98c5076bf3c32994b73 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Fri, 6 Mar 2026 15:50:13 +0000 Subject: [PATCH 7/8] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- pydra/engine/job.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pydra/engine/job.py b/pydra/engine/job.py index 8bec91498..6d3a7d9b2 100644 --- a/pydra/engine/job.py +++ b/pydra/engine/job.py @@ -540,8 +540,8 @@ def _check_for_hash_changes(self): # execution (scalar/pure-Python values are immutable from Pydra's # perspective). Skip the expensive full recomputation in that common case. if not any( - TypeParser.contains_type(FileSet, f.type) or - hasattr(getattr(self.task, f.name), '__bytes_repr__') + TypeParser.contains_type(FileSet, f.type) + or hasattr(getattr(self.task, f.name), "__bytes_repr__") for f in get_fields(self.task) ): logger.debug( From acd0b16af8283c68abaf15785464d771cf001c47 Mon Sep 17 00:00:00 2001 From: Chris Markiewicz Date: Fri, 6 Mar 2026 13:35:02 -0500 Subject: [PATCH 8/8] Apply suggestion from @effigies --- pydra/engine/job.py | 1 - 1 file changed, 1 deletion(-) diff --git a/pydra/engine/job.py b/pydra/engine/job.py index 6d3a7d9b2..6acbb0967 100644 --- a/pydra/engine/job.py +++ b/pydra/engine/job.py @@ -533,7 +533,6 @@ def result(self, return_inputs=False): def _check_for_hash_changes(self): from pydra.utils.typing import TypeParser - from pydra.utils.hash import HasBytesRepr # For tasks whose input fields contain no FileSet types and no values # with custom __bytes_repr__ methods, hashes cannot change during