Merge pull request #246 from PrefectHQ/manual-only

Manual only trigger refactor

docs/generate_docs.py

@@ -43,7 +43,6 @@
             prefect.triggers.all_finished,
             prefect.triggers.manual_only,
             prefect.triggers.always_run,
-            prefect.triggers.never_run,
             prefect.triggers.all_successful,
             prefect.triggers.all_failed,
             prefect.triggers.any_successful,
@@ -140,6 +139,7 @@
             prefect.engine.signals.RETRY,
             prefect.engine.signals.SKIP,
             prefect.engine.signals.DONTRUN,
+            prefect.engine.signals.PAUSE,
         ],
         "title": "Signals",
         "top-level-doc": prefect.engine.signals,

src/prefect/engine/signals.py

@@ -87,3 +87,12 @@ class DONTRUN(PrefectStateSignal):
     Args:
         - message (Any, optional): Defaults to `None`. A message about the signal.
     """
+
+
+class PAUSE(PrefectStateSignal):
+    """
+    Indicates that a task should not run and wait for manual execution.
+
+    Args:
+        - message (Any, optional): Defaults to `None`. A message about the signal.
+    """

src/prefect/engine/task_runner.py

@@ -45,6 +45,11 @@ def inner(self: "TaskRunner", *args: Any, **kwargs: Any) -> State:
             logging.debug("DONTRUN signal raised: {}".format(exc))
             raise
 
+        # PAUSE signals get raised for handling
+        except signals.PAUSE as exc:
+            logging.debug("PAUSE signal raised: {}".format(exc))
+            raise
+
         # RETRY signals are trapped and turned into Retry states
         except signals.RETRY as exc:
             logging.debug("RETRY signal raised")
@@ -166,10 +171,10 @@ def run(
             # a DONTRUN signal at any point breaks the chain and we return
             # the most recently computed state
             except signals.DONTRUN as exc:
-                if "manual_only" in str(exc):
-                    state.cached_inputs = task_inputs or {}
-                    state.message = exc
                 pass
+            except signals.PAUSE as exc:
+                state.cached_inputs = task_inputs or {}
+                state.message = exc
             finally:  # resource is now available
                 for ticket, q in zip(tickets, queues):
                     q.put(ticket)

src/prefect/triggers.py

@@ -29,12 +29,11 @@ def manual_only(upstream_states: Set["state.State"]) -> bool:
 
     Note this doesn't raise a failure, it simply doesn't run the task.
     """
-    raise signals.DONTRUN('Trigger function is "manual_only"')
+    raise signals.PAUSE('Trigger function is "manual_only"')
 
 
 # aliases
 always_run = all_finished
-never_run = manual_only
 
 
 def all_successful(upstream_states: Set["state.State"]) -> bool:

tests/conftest.py

@@ -98,3 +98,27 @@ def clear_registry():
     """
     yield
     prefect.core.registry.REGISTRY.clear()
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--airflow",
+        action="store_true",
+        dest="airflow",
+        help="including this flag will attempt to run airflow compatibility tests",
+    )
+
+
+def pytest_configure(config):
+    config.addinivalue_line(
+        "markers", "airflow: mark test to run only when --airflow flag is provided."
+    )
+
+
+def pytest_runtest_setup(item):
+    mark = item.get_marker("airflow")
+    if mark is not None:
+        if item.config.getoption("--airflow") is False:
+            pytest.mark.skip(
+                "Airflow tests skipped by default unless --airflow flag provided to pytest."
+            )

tests/test_triggers.py

@@ -72,17 +72,17 @@ def test_always_run_with_mixed_states():
 
 
 def test_manual_only_with_all_success():
-    with pytest.raises(signals.DONTRUN):
+    with pytest.raises(signals.PAUSE):
         triggers.manual_only(generate_states(success=3))
 
 
 def test_manual_only_with_all_failed():
-    with pytest.raises(signals.DONTRUN):
+    with pytest.raises(signals.PAUSE):
         triggers.manual_only(generate_states(failed=3))
 
 
 def test_manual_only_with_mixed_states():
-    with pytest.raises(signals.DONTRUN):
+    with pytest.raises(signals.PAUSE):
         triggers.manual_only(generate_states(success=1, failed=1, skipped=1))
 
 

tests/utilities/test_airflow.py

@@ -6,6 +6,7 @@
 from prefect.utilities.airflow_utils import AirFlow
 
 
+@pytest.mark.airflow()
 def test_example_branch_operator():
     flow = AirFlow(dag_id="example_branch_operator")
     res = flow.run(execution_date="2018-09-20", return_tasks=flow.tasks)
@@ -34,6 +35,7 @@ def test_example_branch_operator():
     )
 
 
+@pytest.mark.airflow()
 def test_example_xcom():
     flow = AirFlow(dag_id="example_xcom")
     res = flow.run(execution_date="2018-09-20", return_tasks=flow.tasks)
@@ -51,6 +53,7 @@ def test_example_xcom():
             assert state.result == {"a": "b"}
 
 
+@pytest.mark.airflow()
 def test_example_short_circuit_operator():
     flow = AirFlow(dag_id="example_short_circuit_operator")
     res = flow.run(execution_date="2018-09-20", return_tasks=flow.tasks)
@@ -65,6 +68,7 @@ def test_example_short_circuit_operator():
             assert isinstance(state, Success)
 
 
+@pytest.mark.airflow()
 def test_example_bash_operator():
     flow = AirFlow(dag_id="example_bash_operator")
     res = flow.run(execution_date="2018-09-20", return_tasks=flow.tasks)