Implement task generators and provisional nodes. (#487)

Author: tobiasraabe

docs/source/changes.md

@@ -70,6 +70,7 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 - {pull}`485` adds missing steps to unconfigure pytask after the job is done, which
   caused flaky tests.
 - {pull}`486` adds default names to {class}`~pytask.PPathNode`.
+- {pull}`487` implements task generators and provisional nodes.
 - {pull}`488` raises an error when an invalid value is used in a return annotation.
 - {pull}`489` and {pull}`491` simplifies parsing products and does not raise an error
   when a product annotation is used with the argument name `produces`. And allow

docs/source/how_to_guides/index.md

@@ -19,6 +19,7 @@ capture_warnings
 how_to_influence_build_order
 hashing_inputs_of_tasks
 using_task_returns
+provisional_nodes_and_task_generators
 writing_custom_nodes
 extending_pytask
 the_data_catalog

docs/source/how_to_guides/provisional_nodes_and_task_generators.md

@@ -0,0 +1,94 @@
+# Provisional nodes and task generators
+
+pytask's execution model can usually be separated into three phases.
+
+1. Collection of tasks, dependencies, and products.
+1. Building the DAG.
+1. Executing the tasks.
+
+But, in some situations, pytask needs to be more flexible.
+
+Imagine you want to download a folder with files from an online storage. Before the task
+is completed you do not know the total number of files or their filenames. How can you
+still describe the files as products of the task?
+
+And how would you define another task that depends on these files?
+
+The following sections will explain how you use pytask in these situations.
+
+## Producing provisional nodes
+
+As an example for the aforementioned scenario, let us write a task that downloads all
+files without a file extension from the root folder of the pytask GitHub repository. The
+files are downloaded to a folder called `downloads`. `downloads` is in the same folder
+as the task module because it is a relative path.
+
+```{literalinclude} ../../../docs_src/how_to_guides/provisional_products.py
+---
+emphasize-lines: 4, 22
+---
+```
+
+Since the names of the files are not known when pytask is started, we need to use a
+{class}`~pytask.DirectoryNode` to define the task's product. With a
+{class}`~pytask.DirectoryNode` we can specify where pytask can find the files. The files
+are described with a root path (default is the directory of the task module) and a glob
+pattern (default is `*`).
+
+When we use the {class}`~pytask.DirectoryNode` as a product annotation, we get access to
+the `root_dir` as a {class}`~pathlib.Path` object inside the function, which allows us
+to store the files.
+
+```{note}
+The {class}`~pytask.DirectoryNode` is a provisional node that implements
+{class}`~pytask.PProvisionalNode`. A provisional node is not a {class}`~pytask.PNode`,
+but when its {meth}`~pytask.PProvisionalNode.collect` method is called, it returns
+actual nodes. A {class}`~pytask.DirectoryNode`, for example, returns
+{class}`~pytask.PathNode`.
+```
+
+## Depending on provisional nodes
+
+In the next step, we want to define a task that consumes and merges all previously
+downloaded files into one file.
+
+The difficulty here is how can we reference the downloaded files before they have been
+downloaded.
+
+```{literalinclude} ../../../docs_src/how_to_guides/provisional_task.py
+---
+emphasize-lines: 9
+---
+```
+
+To reference the files that will be downloaded, we use the
+{class}`~pytask.DirectoryNode` is a dependency. Before the task is executed, the list of
+files in the folder defined by the root path and the pattern are automatically collected
+and passed to the task.
+
+If we use a {class}`~pytask.DirectoryNode` with the same `root_dir` and `pattern` in
+both tasks, pytask will automatically recognize that the second task depends on the
+first. If that is not true, you might need to make this dependency more explicit by
+using {func}`@task(after=...) <pytask.task>`, which is explained {ref}`here <after>`.
+
+## Task generators
+
+What if we wanted to process each downloaded file separately instead of dealing with
+them in one task?
+
+For that, we have to write a task generator to define an unknown number of tasks for an
+unknown number of downloaded files.
+
+A task generator is a task function in which we define more tasks, just as if we were
+writing functions in a task module.
+
+The code snippet shows each task takes one of the downloaded files and copies its
+content to a `.txt` file.
+
+```{literalinclude} ../../../docs_src/how_to_guides/provisional_task_generator.py
+```
+
+```{important}
+The generated tasks need to be decoratored with {func}`@task <pytask.task>` to be
+collected.
+```

docs/source/reference_guides/api.md

@@ -190,6 +190,8 @@ Protocols define how tasks and nodes for dependencies and products have to be se
    :show-inheritance:
 .. autoprotocol:: pytask.PTaskWithPath
    :show-inheritance:
+.. autoprotocol:: pytask.PProvisionalNode
+   :show-inheritance:
 ```
 
 ## Nodes
@@ -203,6 +205,8 @@ Nodes are the interface for different kinds of dependencies or products.
    :members:
 .. autoclass:: pytask.PythonNode
    :members:
+.. autoclass:: pytask.DirectoryNode
+   :members:
 ```
 
 To parse dependencies and products from nodes, use the following functions.

docs/source/tutorials/skipping_tasks.md

@@ -13,18 +13,18 @@ skip tasks during development that take too much time to compute right now.
 ```{literalinclude} ../../../docs_src/tutorials/skipping_tasks_example_1.py
 ```
 
-Not only will this task be skipped, but all tasks that depend on
+Not only will this task be skipped, but all tasks depending on
 `time_intensive_product.pkl`.
 
 ## Conditional skipping
 
 In large projects, you may have many long-running tasks that you only want to execute on
-a remote server but not when you are not working locally.
+a remote server, but not when you are not working locally.
 
 In this case, use the {func}`@pytask.mark.skipif <pytask.mark.skipif>` decorator, which
 requires a condition and a reason as arguments.
 
-Place the condition variable in a different module than the task, so you can change it
+Place the condition variable in a module different from the task so you can change it
 without causing a rerun of the task.
 
 ```python

docs_src/how_to_guides/provisional_products.py

@@ -0,0 +1,33 @@
+from pathlib import Path
+
+import httpx
+from pytask import DirectoryNode
+from pytask import Product
+from typing_extensions import Annotated
+
+
+def get_files_without_file_extensions_from_repo() -> list[str]:
+    url = "https://api.github.com/repos/pytask-dev/pytask/git/trees/main"
+    response = httpx.get(url)
+    elements = response.json()["tree"]
+    return [
+        e["path"]
+        for e in elements
+        if e["type"] == "blob" and Path(e["path"]).suffix == ""
+    ]
+
+
+def task_download_files(
+    download_folder: Annotated[
+        Path, DirectoryNode(root_dir=Path("downloads"), pattern="*"), Product
+    ],
+) -> None:
+    """Download files."""
+    # Contains names like CITATION or LICENSE.
+    files_to_download = get_files_without_file_extensions_from_repo()
+
+    for file_ in files_to_download:
+        url = "raw.githubusercontent.com/pytask-dev/pytask/main"
+        response = httpx.get(url=f"{url}/{file_}", timeout=5)
+        content = response.text
+        download_folder.joinpath(file_).write_text(content)

docs_src/how_to_guides/provisional_task.py

@@ -0,0 +1,14 @@
+from pathlib import Path
+
+from pytask import DirectoryNode
+from typing_extensions import Annotated
+
+
+def task_merge_files(
+    paths: Annotated[
+        list[Path], DirectoryNode(root_dir=Path("downloads"), pattern="*")
+    ],
+) -> Annotated[str, Path("all_text.txt")]:
+    """Merge files."""
+    contents = [path.read_text() for path in paths]
+    return "\n".join(contents)

docs_src/how_to_guides/provisional_task_generator.py

@@ -0,0 +1,21 @@
+from pathlib import Path
+
+from pytask import DirectoryNode
+from pytask import task
+from typing_extensions import Annotated
+
+
+@task(is_generator=True)
+def task_copy_files(
+    paths: Annotated[
+        list[Path], DirectoryNode(root_dir=Path("downloads"), pattern="*")
+    ],
+) -> None:
+    """Create tasks to copy each file to a ``.txt`` file."""
+    for path in paths:
+        # The path of the copy will be CITATION.txt, for example.
+        path_to_copy = path.with_suffix(".txt")
+
+        @task
+        def copy_file(path: Annotated[Path, path]) -> Annotated[str, path_to_copy]:
+            return path.read_text()

src/_pytask/collect.py

@@ -30,7 +30,9 @@
 from _pytask.mark_utils import has_mark
 from _pytask.node_protocols import PNode
 from _pytask.node_protocols import PPathNode
+from _pytask.node_protocols import PProvisionalNode
 from _pytask.node_protocols import PTask
+from _pytask.nodes import DirectoryNode
 from _pytask.nodes import PathNode
 from _pytask.nodes import PythonNode
 from _pytask.nodes import Task
@@ -299,6 +301,8 @@ def pytask_collect_task(
             raise ValueError(msg)
 
         path_nodes = Path.cwd() if path is None else path.parent
+
+        # Collect dependencies and products.
         dependencies = parse_dependencies_from_task_function(
             session, path, name, path_nodes, obj
         )
@@ -309,6 +313,9 @@ def pytask_collect_task(
         markers = get_all_marks(obj)
         collection_id = obj.pytask_meta._id if hasattr(obj, "pytask_meta") else None
         after = obj.pytask_meta.after if hasattr(obj, "pytask_meta") else []
+        is_generator = (
+            obj.pytask_meta.is_generator if hasattr(obj, "pytask_meta") else False
+        )
 
         # Get the underlying function to avoid having different states of the function,
         # e.g. due to pytask_meta, in different layers of the wrapping.
@@ -321,7 +328,11 @@ def pytask_collect_task(
                 depends_on=dependencies,
                 produces=products,
                 markers=markers,
-                attributes={"collection_id": collection_id, "after": after},
+                attributes={
+                    "collection_id": collection_id,
+                    "after": after,
+                    "is_generator": is_generator,
+                },
             )
         return Task(
             base_name=name,
@@ -330,41 +341,25 @@ def pytask_collect_task(
             depends_on=dependencies,
             produces=products,
             markers=markers,
-            attributes={"collection_id": collection_id, "after": after},
+            attributes={
+                "collection_id": collection_id,
+                "after": after,
+                "is_generator": is_generator,
+            },
         )
     if isinstance(obj, PTask):
         return obj
     return None
 
 
-_TEMPLATE_ERROR: str = """\
-The provided path of the dependency/product is
-
-{}
-
-, but the path of the file on disk is
-
-{}
-
-Case-sensitive file systems would raise an error because the upper and lower case \
-format of the paths does not match.
-
-Please, align the names to ensure reproducibility on case-sensitive file systems \
-(often Linux or macOS) or disable this error with 'check_casing_of_paths = false' in \
-the pyproject.toml file.
-
-Hint: If parts of the path preceding your project directory are not properly \
-formatted, check whether you need to call `.resolve()` on `SRC`, `BLD` or other paths \
-created from the `__file__` attribute of a module.
-"""
-
-
 _TEMPLATE_ERROR_DIRECTORY: str = """\
 The path '{path}' points to a directory, although only files are allowed."""
 
 
 @hookimpl(trylast=True)
-def pytask_collect_node(session: Session, path: Path, node_info: NodeInfo) -> PNode:  # noqa: C901, PLR0912
+def pytask_collect_node(  # noqa: C901, PLR0912
+    session: Session, path: Path, node_info: NodeInfo
+) -> PNode | PProvisionalNode:
     """Collect a node of a task as a :class:`pytask.PNode`.
 
     Strings are assumed to be paths. This might be a strict assumption, but since this
@@ -384,6 +379,21 @@ def pytask_collect_node(session: Session, path: Path, node_info: NodeInfo) -> PN
     """
     node = node_info.value
 
+    if isinstance(node, DirectoryNode):
+        if node.root_dir is None:
+            node.root_dir = path
+        if (
+            not node.name
+            or node.name == node.root_dir.joinpath(node.pattern).as_posix()
+        ):
+            short_root_dir = shorten_path(
+                node.root_dir, session.config["paths"] or (session.config["root"],)
+            )
+            node.name = Path(short_root_dir, node.pattern).as_posix()
+
+    if isinstance(node, PProvisionalNode):
+        return node
+
     if isinstance(node, PythonNode):
         node.node_info = node_info
         if not node.name:
@@ -418,9 +428,11 @@ def pytask_collect_node(session: Session, path: Path, node_info: NodeInfo) -> PN
         raise ValueError(_TEMPLATE_ERROR_DIRECTORY.format(path=node.path))
 
     if isinstance(node, PNode):
+        if not node.name:
+            node.name = create_name_of_python_node(node_info)
         return node
 
-    if isinstance(node, UPath):
+    if isinstance(node, UPath):  # pragma: no cover
         if not node.protocol:
             node = Path(node)
         else:
@@ -459,6 +471,28 @@ def pytask_collect_node(session: Session, path: Path, node_info: NodeInfo) -> PN
     return PythonNode(value=node, name=node_name, node_info=node_info)
 
 
+_TEMPLATE_ERROR: str = """\
+The provided path of the dependency/product is
+
+{}
+
+, but the path of the file on disk is
+
+{}
+
+Case-sensitive file systems would raise an error because the upper and lower case \
+format of the paths does not match.
+
+Please, align the names to ensure reproducibility on case-sensitive file systems \
+(often Linux or macOS) or disable this error with 'check_casing_of_paths = false' in \
+your pytask configuration file.
+
+Hint: If parts of the path preceding your project directory are not properly \
+formatted, check whether you need to call `.resolve()` on `SRC`, `BLD` or other paths \
+created from the `__file__` attribute of a module.
+"""
+
+
 def _raise_error_if_casing_of_path_is_wrong(
     path: Path, check_casing_of_paths: bool
 ) -> None: