Implement a conditional skip marker

docs/tutorials/how_to_skip_tasks.rst

@@ -0,0 +1,59 @@
+How to skip tasks
+==================
+
+Tasks are skipped automatically if neither their file nor any of their dependencies have
+changed and all products exist.
+
+In addition, you may want pytask to skip tasks either generally or if certain conditions
+are fulfilled. Skipping means the task itself and all tasks that depend on it will not
+be executed, even if the task file or their dependencies have changed or products are
+missing.
+
+This can be useful for example if you are working on a task that creates the dependency
+of a long running task and you are not interested in the long running task's product
+for the moment. In that case you can simply use ``@pytask.mark.skip`` in front of the
+long running task to stop it from running:
+
+.. code-block:: python
+
+    # Content of task_create_dependency.py
+
+    @pytask.mark.produces("dependency_of_long_running_task.md")
+    def task_you_are_working_on(produces):
+        ...
+
+
+    # Content of task_long_running.py
+
+    @pytask.mark.skip
+    @pytask.mark.depends_on("dependency_of_long_running_task.md")
+    def task_that_takes_really_long_to_run(depends_on):
+
+
+In large projects, you may have many long running tasks that you only want to
+execute sporadically, e.g. when you are not working locally but running the project
+on a server.
+
+In that case, we recommend using ``@pytask.mark.skip_if`` which lets you supply a
+condition and a reason as arguments:
+
+
+.. code-block:: python
+
+    # Content of a config.py
+
+    FAST_FLAG = True
+
+    # Content of task_create_dependency.py
+
+    @pytask.mark.produces("run_always.md")
+    def task_always(produces):
+        ...
+
+    # Content of task_long_running.py
+
+    from config import FAST_FLAG
+
+    @pytask.mark.skipif(FAST_FLAG, "Fast Flag active")
+    @pytask.mark.depends_on("dependency_of_long_running_task.md")
+    def task_that_takes_really_long_to_run(depends_on):

docs/tutorials/index.rst

@@ -17,6 +17,7 @@ organize and start your own project.
    how_to_select_tasks
    how_to_clean
    how_to_collect
+   how_to_skip_tasks
    how_to_make_tasks_persist
    how_to_capture
    how_to_invoke_pytask

src/_pytask/skipping.py

@@ -16,6 +16,11 @@ def skip_ancestor_failed(reason: str = "No reason provided.") -> str:
     return reason
 
 
+def skip_if(condition: bool, reason: str = "No reason provided.") -> tuple:
+    """Function to parse information in ``@pytask.mark.skip_if``."""
+    return condition, reason
+
+
 @hookimpl
 def pytask_parse_config(config):
     markers = {
@@ -24,6 +29,8 @@ def pytask_parse_config(config):
         "failed.",
         "skip_unchanged": "Internal decorator applied to tasks which have already been "
         "executed and have not been changed.",
+        "skip_if": "Skip a task and all its subsequent tasks in case a condition is "
+        "fulfilled.",
     }
     config["markers"] = {**config["markers"], **markers}
 
@@ -46,6 +53,14 @@ def pytask_execute_task_setup(task):
     if markers:
         raise Skipped
 
+    markers = get_specific_markers_from_task(task, "skip_if")
+    if markers:
+        marker_args = [skip_if(*marker.args, **marker.kwargs) for marker in markers]
+        message = "\n".join([arg[1] for arg in marker_args])
+        should_skip = all(arg[0] for arg in marker_args)
+        if should_skip:
+            raise Skipped(message)
+
 
 @hookimpl
 def pytask_execute_task_process_report(session, report):

tests/test_skipping.py

@@ -102,6 +102,66 @@ def task_second():
     assert isinstance(session.execution_reports[1].exc_info[1], Skipped)
 
 
+@pytest.mark.end_to_end
+def test_if_skip_if_decorator_is_applied_skipping(tmp_path):
+    source = """
+    import pytask
+
+    @pytask.mark.skip_if(condition=True, reason="bla")
+    @pytask.mark.produces("out.txt")
+    def task_first():
+        assert False
+
+    @pytask.mark.depends_on("out.txt")
+    def task_second():
+        assert False
+    """
+    tmp_path.joinpath("task_dummy.py").write_text(textwrap.dedent(source))
+
+    session = main({"paths": tmp_path})
+    node = session.collection_reports[0].node
+    assert len(node.markers) == 1
+    assert node.markers[0].name == "skip_if"
+    assert node.markers[0].args == ()
+    assert node.markers[0].kwargs == {"condition": True, "reason": "bla"}
+
+    assert session.execution_reports[0].success
+    assert isinstance(session.execution_reports[0].exc_info[1], Skipped)
+    assert session.execution_reports[1].success
+    assert isinstance(session.execution_reports[1].exc_info[1], Skipped)
+    assert session.execution_reports[0].exc_info[1].args[0] == "bla"
+
+
+@pytest.mark.end_to_end
+def test_if_skip_if_decorator_is_applied_execute(tmp_path):
+    source = """
+    import pytask
+
+    @pytask.mark.skip_if(False)
+    @pytask.mark.produces("out.txt")
+    def task_first(produces):
+        with open(produces, "w") as f:
+            f.write("hello world.")
+
+    @pytask.mark.depends_on("out.txt")
+    def task_second():
+        pass
+    """
+    tmp_path.joinpath("task_dummy.py").write_text(textwrap.dedent(source))
+
+    session = main({"paths": tmp_path})
+    node = session.collection_reports[0].node
+
+    assert len(node.markers) == 1
+    assert node.markers[0].name == "skip_if"
+    assert node.markers[0].args == (False,)
+    assert node.markers[0].kwargs == {}
+    assert session.execution_reports[0].success
+    assert session.execution_reports[0].exc_info is None
+    assert session.execution_reports[1].success
+    assert session.execution_reports[1].exc_info is None
+
+
 @pytest.mark.unit
 @pytest.mark.parametrize(
     ("marker_name", "expectation"),