Can you combine the functional interface with complex task repetition?

[nkruskamp]

Hi!

I have a use case where I'd like to do complex task repetition, and I would like to do so in the functional interface (interactive jupyter session). I was not able to find an example of this in the documentation. I am wondering if it is possible to combine the two.

https://pytask-dev.readthedocs.io/en/stable/tutorials/repeating_tasks_with_different_inputs.html#complex-example

Here is a basic example as I would approach it taken from the complex example in the documentation above:

pytask.build(
    tasks=[task_create_random_data(**kwargs) for kwargs in ID_TO_KWARGS.values()],
    verbose=10,
    marker_expression="persist",
)

As it is written, it initiated the functional interface, but ran the function iterations outside of pytask and pytask reported no tasks. Not the intended result!

Can you please advise of if this is possible and the correct way to approach this? Thanks!

[tobiasraabe]

Hi @nkruskamp, here is the documentation you are looking for https://pytask-dev.readthedocs.io/en/v0.5.6/how_to_guides/functional_interface.html#repeated-tasks.

While drafting it I found a bug that is fixed in v0.5.6. Happy coding!

[nkruskamp]

@tobiasraabe thanks so much for this! I wanted to clarify some things, and maybe this has now uncovered another bug and I should open an issue for it, but I will describe here as well.

Info
I have updated to the latest version 0.5.6 for these tests.
Windows 11
VS Code 1.105.1
Jupyter Extension 2025.9.1. Interactive Session.
Python version 3.12.12
Conda environment
pytask version 0.5.6

Clarifying questions:

1. in the documentation, the text says to use the @task decorator in the loop, but the example code does not have that. Can you say if using the decorator is necessary?
2. Related, I see in the code example you are passing the i indexer directly to the function parameters instead of using the task decorator with @task(id=_id, kwargs=kwargs) to pass arguments to the function. I have been using kwargs as the parameters to functions I am writing are often complex sets of inputs.

Issue

I am running the example code as written out of a file C:/Users/{username}/Downloads/pytask_test.py.
The output path in the function has been specified Path(f"C:/Users/{username}/Downloads/output_{i}.txt").

With dry_run=True, I see Collected 3 tasks, however in the summary I see 1 Would be executed when I expect to see a 3 there.

with dry_run=False, I again see Collected 3 tasks, but I then get an error. However, output_2.txt is created, not 0 or 1. If I change the range, the last file always gets created and the below error still happens.

╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
│ c:\Users\{username}\AppData\Local\miniforge3\envs\{env}\Lib\pathlib.py:840 in stat               │
│                                                                                                  │
│    837 │   │   Return the result of the stat() system call on this path, like                    │
│    838 │   │   os.stat() does.                                                                   │
│    839 │   │   """                                                                               │
│ ❱  840 │   │   return os.stat(self, follow_symlinks=follow_symlinks)                             │
│    841 │                                                                                         │
│    842 │   def lstat(self):                                                                      │
│    843 │   │   """                                                                               │
│                                                                                                  │
│ ╭─────────────────────────────────────────── locals ───────────────────────────────────────────╮ │
│ │ follow_symlinks = True                                                                       │ │
│ │            self = WindowsPath('C:/Users/{username}/Downloads/<ipython-input-1-da7f262f1349>')│ │
│ ╰──────────────────────────────────────────────────────────────────────────────────────────────╯ │
╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
OSError: [WinError 123] The filename, directory name, or volume label syntax is incorrect: 
'C:\\Users\\{username}\\Downloads\\<ipython-input-1-da7f262f1349>'

If I pass marker_expression="persist", to build, it will correctly skip the 1 file that gets created and no error will occur, but no other files will be created either.

That's as far as I've made it for now. Please let me know if there any other info I can pass along and if you'd like me to open an issue for this. Thanks again for your help!

[tobiasraabe]

Thanks for the follow-up!

1. I clarified the docs in Clarify documentation. #721
2. If you use the @task decorator, you can pass args via kwargs, but you can also use other means: https://pytask-dev.readthedocs.io/en/stable/how_to_guides/interfaces_for_dependencies_products.html.

Regarding the issue, please, post a standalone script and the command to run it as I have not been able to reproduce the issue from your description.

[nkruskamp]

Thanks! I am just running examples from the documentation to get the error in the interactive output I shared above. The full script I am running is below.

I am testing these in both interactive sessions and running on the CLI. CLI works as expected when tasks are decorated correctly. Interestingly, when the task loop is NOT decorated correctly, it still picks up the last task (`output_3.txt') from that loop and runs it in CLI.

# %%
from pathlib import Path
from typing import Annotated

import pytask
from pytask import task, Product


# %%
def task_create_first_file() -> Annotated[str, Path("first.txt")]:
    return "Hello, "


task_create_second_file = task(
    name="task_create_second_file", produces=Path("second.txt")
)(lambda *x: "World!")


def task_merge_files(
    first: Path = Path("first.txt"), second: Path = Path("second.txt")
) -> Annotated[str, Path("hello_world.txt")]:
    return first.read_text() + second.read_text()


# task_create_first_file() and task_create_second_file() will be shown as failed
# task_merge_files() will be skipped due to previous failures.
session = pytask.build(
    tasks=[task_create_first_file, task_merge_files, task_create_second_file],
)

# %%
# Create repeated tasks with different parameters
tasks = []
for i in range(3):
    # @task # testing with CLI
    # NOTE: On CLI when this is not used, the first two tasks will not run, but the
    # third one will. pytask will collect only 4 tasks (1 here and 3 from above) tasks.
    # NOTE: on CLI when this is used, all three tasks will run. pytask will collect 6
    # tasks (3 here and 3 from above).
    def task_create_file(
        value: int = i * 100, path: Annotated[Path, Product] = Path(f"output_{i}.txt")
    ) -> None:
        path.write_text(f"Result: {value}")

    tasks.append(task_create_file)

# the last task output_2 will be successful but success output will not be shown.
# output_0 and output_1 will fail with symlink issue
session = pytask.build(tasks=tasks, check_casing_of_paths=False)

# %%
# Cleanup
# for name in ("first.txt", "second.txt", "hello_world.txt"):
#     Path(name).unlink(missing_ok=True)

# for i in range(3):
#     Path(f"output_{i}.txt").unlink(missing_ok=True)
# %%

[nkruskamp]

@tobiasraabe I think I realized one important thing!

I am not running this code in .ipynb files. I test .py files against an interactive session in VS Code using the Jupyter extension. You might notice the lines in the code above where "cells" are marked. This type of execution is when this error occurs.

When the above code is put into an .ipynb file, the code runs as expected, with the added parameter check_casing_of_paths=False.