Merge branch 'main' into delayed-nodes

Author: tobiasraabe

.github/workflows/main.yml

@@ -5,9 +5,6 @@ concurrency:
   group: ${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
 
-env:
-  CONDA_EXE: mamba
-
 on:
   push:
     branches:

.pre-commit-config.yaml

@@ -33,18 +33,18 @@ repos:
     hooks:
     -   id: sort-all
 -   repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.3.1
+    rev: v0.3.2
     hooks:
       - id: ruff-format
       - id: ruff
         args: [--unsafe-fixes]
 -   repo: https://github.com/dosisod/refurb
-    rev: v1.28.0
+    rev: v2.0.0
     hooks:
     -   id: refurb
         args: [--ignore, FURB126]
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.8.0
+    rev: v1.9.0
     hooks:
     -   id: mypy
         additional_dependencies: [
@@ -80,8 +80,6 @@ repos:
 -   repo: https://github.com/nbQA-dev/nbQA
     rev: 1.8.4
     hooks:
-    -   id: nbqa-black
-    -   id: nbqa-isort
     -   id: nbqa-mypy
         args: [--ignore-missing-imports]
 -   repo: https://github.com/kynan/nbstripout

docs/source/changes.md

@@ -20,6 +20,8 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 - {pull}`566` makes universal-pathlib an official dependency.
 - {pull}`568` restricts `task_files` to a list of patterns and raises a better error.
 - {pull}`569` removes the hooks related to the creation of the DAG.
+- {pull}`571` removes redundant calls to `PNode.state()` which causes a high penalty for
+  remote files.
 
 ## 0.4.5 - 2024-01-09
 

src/_pytask/_hashlib.py

@@ -232,5 +232,5 @@ def hash_value(value: Any) -> int | str:
     if isinstance(value, str):
         value = value.encode()
     if isinstance(value, bytes):
-        return str(hashlib.sha256(value).hexdigest())
+        return hashlib.sha256(value).hexdigest()
     return hash(value)

src/_pytask/data_catalog.py

@@ -100,7 +100,7 @@ def add(self, name: str, node: PNode | PProvisionalNode | None = None) -> None:
             raise TypeError(msg)
 
         if node is None:
-            filename = str(hashlib.sha256(name.encode()).hexdigest())
+            filename = hashlib.sha256(name.encode()).hexdigest()
             if isinstance(self.default_node, PPathNode):
                 self.entries[name] = self.default_node(
                     name=name, path=self.path / f"{filename}.pkl"

src/_pytask/database_utils.py

@@ -69,11 +69,10 @@ def update_states_in_database(session: Session, task_signature: str) -> None:
         _create_or_update_state(task_signature, node.signature, hash_)
 
 
-def has_node_changed(task: PTask, node: PTask | PNode) -> bool:
+def has_node_changed(task: PTask, node: PTask | PNode, state: str | None) -> bool:
     """Indicate whether a single dependency or product has changed."""
     # If node does not exist, we receive None.
-    node_state = node.state()
-    if node_state is None:
+    if state is None:
         return True
 
     with DatabaseSession() as session:
@@ -83,4 +82,4 @@ def has_node_changed(task: PTask, node: PTask | PNode) -> bool:
     if db_state is None:
         return True
 
-    return node_state != db_state.hash_
+    return state != db_state.hash_

src/_pytask/execute.py

@@ -146,8 +146,8 @@ def pytask_execute_task_setup(session: Session, task: PTask) -> None:  # noqa: C
             node = dag.nodes[node_signature].get("task") or dag.nodes[
                 node_signature
             ].get("node")
-
-            if node_signature in predecessors and not node.state():
+            node_state = node.state()
+            if node_signature in predecessors and not node_state:
                 msg = f"{task.name!r} requires missing node {node.name!r}."
                 if IS_FILE_SYSTEM_CASE_SENSITIVE:
                     msg += (
@@ -162,7 +162,7 @@ def pytask_execute_task_setup(session: Session, task: PTask) -> None:  # noqa: C
             ):
                 continue
 
-            has_changed = has_node_changed(task=task, node=node)
+            has_changed = has_node_changed(task=task, node=node, state=node_state)
             if has_changed:
                 needs_to_be_executed = True
                 break

src/_pytask/nodes.py

@@ -5,6 +5,7 @@
 import hashlib
 import inspect
 import pickle
+from contextlib import suppress
 from os import stat_result
 from pathlib import Path  # noqa: TCH003
 from typing import TYPE_CHECKING
@@ -83,12 +84,10 @@ def signature(self) -> str:
 
     def state(self) -> str | None:
         """Return the state of the node."""
-        try:
+        with suppress(OSError):
             source = inspect.getsource(self.function)
-        except OSError:
-            return None
-        else:
             return hashlib.sha256(source.encode()).hexdigest()
+        return None
 
     def execute(self, **kwargs: Any) -> Any:
         """Execute the task."""
@@ -145,10 +144,7 @@ def signature(self) -> str:
 
     def state(self) -> str | None:
         """Return the state of the node."""
-        if self.path.exists():
-            modification_time = self.path.stat().st_mtime
-            return hash_path(self.path, modification_time)
-        return None
+        return _get_state(self.path)
 
     def execute(self, **kwargs: Any) -> Any:
         """Execute the task."""
@@ -188,9 +184,7 @@ def state(self) -> str | None:
         The state is given by the modification timestamp.
 
         """
-        if self.path.exists():
-            return _get_state(self.path)
-        return None
+        return _get_state(self.path)
 
     def load(self, is_product: bool = False) -> Path:  # noqa: ARG002
         """Load the value."""
@@ -324,9 +318,7 @@ def from_path(cls, path: Path) -> PickleNode:
         return cls(name=path.as_posix(), path=path)
 
     def state(self) -> str | None:
-        if self.path.exists():
-            return _get_state(self.path)
-        return None
+        return _get_state(self.path)
 
     def load(self, is_product: bool = False) -> Any:
         if is_product:
@@ -377,15 +369,19 @@ def collect(self) -> list[Path]:
         return list(self.root_dir.glob(self.pattern))  # type: ignore[union-attr]
 
 
-def _get_state(path: Path) -> str:
+def _get_state(path: Path) -> str | None:
     """Get state of a path.
 
     A simple function to handle local and remote files.
 
     """
-    stat = path.stat()
+    try:
+        stat = path.stat()
+    except FileNotFoundError:
+        return None
+
     if isinstance(stat, stat_result):
-        modification_time = path.stat().st_mtime
+        modification_time = stat.st_mtime
         return hash_path(path, modification_time)
     if isinstance(stat, UPathStatResult):
         return stat.as_info().get("ETag", "0")

src/_pytask/persist.py

@@ -40,21 +40,25 @@ def pytask_execute_task_setup(session: Session, task: PTask) -> None:
 
     """
     if has_mark(task, "persist"):
-        all_nodes_exist = all(
+        all_states = [
             (
                 session.dag.nodes[name].get("task") or session.dag.nodes[name]["node"]
             ).state()
             for name in node_and_neighbors(session.dag, task.signature)
-        )
+        ]
+        all_nodes_exist = all(all_states)
 
         if all_nodes_exist:
             any_node_changed = any(
                 has_node_changed(
                     task=task,
                     node=session.dag.nodes[name].get("task")
                     or session.dag.nodes[name]["node"],
+                    state=state,
+                )
+                for name, state in zip(
+                    node_and_neighbors(session.dag, task.signature), all_states
                 )
-                for name in node_and_neighbors(session.dag, task.signature)
             )
             if any_node_changed:
                 collect_provisional_products(session, task)

tests/conftest.py

@@ -54,7 +54,7 @@ class SysPathsSnapshot:
     """A snapshot for sys.path."""
 
     def __init__(self) -> None:
-        self.__saved = list(sys.path), list(sys.meta_path)
+        self.__saved = sys.path.copy(), sys.meta_path.copy()
 
     def restore(self) -> None:
         sys.path[:], sys.meta_path[:] = self.__saved
@@ -64,7 +64,7 @@ class SysModulesSnapshot:
     """A snapshot for sys.modules."""
 
     def __init__(self) -> None:
-        self.__saved = dict(sys.modules)
+        self.__saved = sys.modules.copy()
 
     def restore(self) -> None:
         sys.modules.clear()