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,14 @@ 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.
+- {pull}`573` removes the `pytask_execute_create_scheduler` hook.
+
+## 0.4.6 - 2024-03-13
+
+- {pull}`576` fixes accidentally collecting `pytask.MarkGenerator` when using
+  `from pytask import mark`.
 
 ## 0.4.5 - 2024-01-09
 

docs/source/reference_guides/hookspecs.md

@@ -69,7 +69,6 @@ The following hooks execute the tasks and log information on the result in the t
 ```{eval-rst}
 .. autofunction:: pytask_execute
 .. autofunction:: pytask_execute_log_start
-.. autofunction:: pytask_execute_create_scheduler
 .. autofunction:: pytask_execute_build
 .. autofunction:: pytask_execute_task_protocol
 .. autofunction:: pytask_execute_task_log_start

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

@@ -60,7 +60,7 @@ def pytask_post_parse(config: dict[str, Any]) -> None:
 def pytask_execute(session: Session) -> None:
     """Execute tasks."""
     session.hook.pytask_execute_log_start(session=session)
-    session.scheduler = session.hook.pytask_execute_create_scheduler(session=session)
+    session.scheduler = TopologicalSorter.from_dag(session.dag)
     session.hook.pytask_execute_build(session=session)
     session.hook.pytask_execute_log_end(
         session=session, reports=session.execution_reports
@@ -76,12 +76,6 @@ def pytask_execute_log_start(session: Session) -> None:
     console.print()
 
 
-@hookimpl(trylast=True)
-def pytask_execute_create_scheduler(session: Session) -> TopologicalSorter:
-    """Create a scheduler based on topological sorting."""
-    return TopologicalSorter.from_dag(session.dag)
-
-
 @hookimpl
 def pytask_execute_build(session: Session) -> bool | None:
     """Execute tasks."""
@@ -146,8 +140,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 +156,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/hookspecs.py

@@ -233,16 +233,6 @@ def pytask_execute_log_start(session: Session) -> None:
     """
 
 
-@hookspec(firstresult=True)
-def pytask_execute_create_scheduler(session: Session) -> Any:
-    """Create a scheduler for the execution.
-
-    The scheduler provides information on which tasks are able to be executed. Its
-    foundation is likely a topological ordering of the tasks based on the DAG.
-
-    """
-
-
 @hookspec(firstresult=True)
 def pytask_execute_build(session: Session) -> Any:
     """Execute the build.

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")