Repository: PrefectHQ/prefect

Issue:
Change 'FlowRunTask' to 'StartFlowRunTask'
## Current behavior
<!-- Please describe how the feature works today -->

The FlowRunTask is a handy task commonly used by our community, however new users may be confused to it's functionality and add it to their flows without realizing they are calling for another FlowRun.


## Proposed behavior
<!-- Please describe your proposed change to the current behavior -->

This proposed change is purely semantic, as some users may be confused with the functionality of FlowRunTask. Now that we have [CancelFlowRunTask](https://docs.prefect.io/api/latest/tasks/prefect.html#cancelflowruntask) and [RenameFlowRunTask](https://docs.prefect.io/api/latest/tasks/prefect.html#renameflowruntask), changing the name of FlowRunTask to **StartFlowRunTask** will provide more consistency across Prefect Task Library tasks and potentially reduce new user confusion in the future. A minor change that could potentially affect many users, but can provide clarity in the long-term for these new users.


## Benefits
<!-- Please give an example of how the enhancement would be useful -->

Clearly discern functionality of the task by name.
Provide consistency across Prefect Task Library tasks.
Potentially reduce debugging confusion if multiple instances of a flow are scheduled by the Agent (a non-user).

Thanks!

Discussion:

Patch:
diff --git a/changes/pr3539.yaml b/changes/pr3539.yaml
new file mode 100644
index 0000000000..b8171f699a
--- /dev/null
+++ b/changes/pr3539.yaml
@@ -0,0 +1,2 @@
+deprecation:
+  - "`FlowRunTask`, `RenameFlowRunTask`, and `CancelFlowRunTask` have been renamed to `StartFlowRun`, `RenameFlowRun`, and `CancelFlowRun` respectively - [#3539](https://github.com/PrefectHQ/prefect/pull/3539)"
diff --git a/docs/core/idioms/flow-to-flow.md b/docs/core/idioms/flow-to-flow.md
index afaaccb09b..de44d0cbf9 100644
--- a/docs/core/idioms/flow-to-flow.md
+++ b/docs/core/idioms/flow-to-flow.md
@@ -4,7 +4,7 @@ There are many situations where users want to configure dependencies at the flow
 if you have a data preprocessing flow that you maintain separately from a model training flow, and you 
 always want to ensure that the preprocessing flow runs before the training flow.
 
-To support this, Prefect provides a convenient [built-in task for creating Flow runs](/api/latest/tasks/prefect.html#flowruntask)
+To support this, Prefect provides a convenient [built-in task for creating Flow runs](/api/latest/tasks/prefect.html#startflowrun)
 through the Prefect API.  This task supports:
 - triggering flow runs when certain conditions are met in another flow
 - providing parameter values that are generated at runtime by another flow
@@ -17,14 +17,14 @@ through the Prefect API.  This task supports:
 
 Let's say that you want to always run a flow with parameters that are generated by another flow. 
 Naively, you might manually wait for one flow to finish and then manually trigger the next flow to run
-with the appropriate parameter values.  Prefect makes this pattern easy to automate via the `FlowRunTask`:
+with the appropriate parameter values.  Prefect makes this pattern easy to automate via the `StartFlowRun`:
 
 
 :::: tabs
 ::: tab "Functional API"
 ```python
 from prefect import task, Flow
-from prefect.tasks.prefect import FlowRunTask
+from prefect.tasks.prefect import StartFlowRun
 
 
 @task
@@ -33,7 +33,7 @@ def extract_some_data():
 
 
 # assumes you have registered a flow named "example" in a project named "examples"
-flow_run = FlowRunTask(flow_name="example", project_name="examples")
+flow_run = StartFlowRun(flow_name="example", project_name="examples")
 
 with Flow("parent-flow") as flow:
     flow_run(parameters=extract_some_data)
@@ -43,7 +43,7 @@ with Flow("parent-flow") as flow:
 ::: tab "Imperative API"
 ```python
 from prefect import Task, Flow
-from prefect.tasks.prefect import FlowRunTask
+from prefect.tasks.prefect import StartFlowRun
 
 
 class ExtractSomeData(Task)
@@ -53,7 +53,7 @@ class ExtractSomeData(Task)
 extract_some_data = ExtractSomeData(name="extract_some_data")
 
 # assumes you have registered a flow named "exampled" in a project named "examples"
-flow_run = FlowRunTask(flow_name="example", project_name="examples")
+flow_run = StartFlowRun(flow_name="example", project_name="examples")
 
 with Flow("parent-flow") as flow:
     flow_run.set_upstream(extract_some_data, key="parameters")
@@ -77,7 +77,7 @@ The following example creates the following Flow-of-Flows that runs every weekda
 ```python
 from prefect import Flow
 from prefect.schedules import CronSchedule
-from prefect.tasks.prefect import FlowRunTask
+from prefect.tasks.prefect import StartFlowRun
 
 
 weekday_schedule = CronSchedule(
@@ -86,10 +86,10 @@ weekday_schedule = CronSchedule(
 
 
 # assumes you have registered the following flows in a project named "examples"
-flow_a = FlowRunTask(flow_name="A", project_name="examples", wait=True)
-flow_b = FlowRunTask(flow_name="B", project_name="examples", wait=True)
-flow_c = FlowRunTask(flow_name="C", project_name="examples", wait=True)
-flow_d = FlowRunTask(flow_name="D", project_name="examples", wait=True)
+flow_a = StartFlowRun(flow_name="A", project_name="examples", wait=True)
+flow_b = StartFlowRun(flow_name="B", project_name="examples", wait=True)
+flow_c = StartFlowRun(flow_name="C", project_name="examples", wait=True)
+flow_d = StartFlowRun(flow_name="D", project_name="examples", wait=True)
 
 with Flow("parent-flow", schedule=weekday_schedule) as flow:
     b = flow_b(upstream_tasks=[flow_a])
@@ -102,7 +102,7 @@ with Flow("parent-flow", schedule=weekday_schedule) as flow:
 ```python
 from prefect import Flow
 from prefect.schedules import CronSchedule
-from prefect.tasks.prefect import FlowRunTask
+from prefect.tasks.prefect import StartFlowRun
 
 
 weekday_schedule = CronSchedule(
@@ -111,10 +111,10 @@ weekday_schedule = CronSchedule(
 
 
 # assumes you have registered the following flows in a project named "examples"
-flow_a = FlowRunTask(flow_name="A", project_name="examples", wait=True)
-flow_b = FlowRunTask(flow_name="B", project_name="examples", wait=True)
-flow_c = FlowRunTask(flow_name="C", project_name="examples", wait=True)
-flow_d = FlowRunTask(flow_name="D", project_name="examples", wait=True)
+flow_a = StartFlowRun(flow_name="A", project_name="examples", wait=True)
+flow_b = StartFlowRun(flow_name="B", project_name="examples", wait=True)
+flow_c = StartFlowRun(flow_name="C", project_name="examples", wait=True)
+flow_d = StartFlowRun(flow_name="D", project_name="examples", wait=True)
 
 with Flow("parent-flow", schedule=weekday_schedule) as flow:
     flow_b.set_upstream(flow_a)
diff --git a/docs/outline.toml b/docs/outline.toml
index f695cc8e2f..9631c5204f 100644
--- a/docs/outline.toml
+++ b/docs/outline.toml
@@ -276,7 +276,7 @@ classes = ["DatasetCreateFromDelimitedFiles",
 [pages.tasks.prefect]
 title = "Prefect Tasks"
 module = "prefect.tasks.prefect"
-classes = ["FlowRunTask", "RenameFlowRunTask", "CancelFlowRunTask"]
+classes = ["StartFlowRun", "RenameFlowRun", "CancelFlowRun"]
 
 [pages.tasks.github]
 title = "GitHub Tasks"
diff --git a/src/prefect/tasks/prefect/__init__.py b/src/prefect/tasks/prefect/__init__.py
index af62fb115a..49d4256a04 100644
--- a/src/prefect/tasks/prefect/__init__.py
+++ b/src/prefect/tasks/prefect/__init__.py
@@ -2,6 +2,6 @@
 Tasks for interacting with the Prefect API
 """
 
-from prefect.tasks.prefect.flow_run import FlowRunTask
-from prefect.tasks.prefect.flow_run_rename import RenameFlowRunTask
-from prefect.tasks.prefect.flow_run_cancel import CancelFlowRunTask
+from prefect.tasks.prefect.flow_run import FlowRunTask, StartFlowRun
+from prefect.tasks.prefect.flow_run_rename import RenameFlowRunTask, RenameFlowRun
+from prefect.tasks.prefect.flow_run_cancel import CancelFlowRunTask, CancelFlowRun
diff --git a/src/prefect/tasks/prefect/flow_run.py b/src/prefect/tasks/prefect/flow_run.py
index 96a8b63110..1f48cd5103 100644
--- a/src/prefect/tasks/prefect/flow_run.py
+++ b/src/prefect/tasks/prefect/flow_run.py
@@ -1,4 +1,5 @@
 import time
+import warnings
 from typing import Any
 
 from prefect import context, Task
@@ -8,7 +9,7 @@ from prefect.utilities.graphql import EnumValue, with_args
 from prefect.utilities.tasks import defaults_from_attrs
 
 
-class FlowRunTask(Task):
+class StartFlowRun(Task):
     """
     Task used to kick off a flow run using Prefect Core's server or Prefect Cloud.  If multiple
     versions of the flow are found, this task will kick off the most recent unarchived version.
@@ -88,9 +89,9 @@ class FlowRunTask(Task):
 
         Example:
             ```python
-            from prefect.tasks.prefect.flow_run import FlowRunTask
+            from prefect.tasks.prefect.flow_run import StartFlowRun
 
-            kickoff_task = FlowRunTask(project_name="Hello, World!", flow_name="My Cloud Flow")
+            kickoff_task = StartFlowRun(project_name="Hello, World!", flow_name="My Cloud Flow")
             ```
 
         """
@@ -162,3 +163,13 @@ class FlowRunTask(Task):
                     f"{flow_run_id} finished in state {flow_run_state}"
                 )
                 raise exc
+
+
+class FlowRunTask(StartFlowRun):
+    def __new__(cls, *args, **kwargs):  # type: ignore
+        warnings.warn(
+            "`FlowRunTask` has been renamed to `prefect.tasks.prefect.StartFlowRun`,"
+            "please update your code accordingly",
+            stacklevel=2,
+        )
+        return super().__new__(cls)
diff --git a/src/prefect/tasks/prefect/flow_run_cancel.py b/src/prefect/tasks/prefect/flow_run_cancel.py
index bdf8b1404d..d74fe93378 100644
--- a/src/prefect/tasks/prefect/flow_run_cancel.py
+++ b/src/prefect/tasks/prefect/flow_run_cancel.py
@@ -1,11 +1,13 @@
-from prefect import Task
-from prefect.client import Client
+import warnings
 from typing import Any
+
 import prefect
+from prefect import Task
+from prefect.client import Client
 from prefect.utilities.tasks import defaults_from_attrs
 
 
-class CancelFlowRunTask(Task):
+class CancelFlowRun(Task):
     """
     Task to cancel a flow run. If `flow_run_id` is not provided,
     `flow_run_id` from `prefect.context` will be used by default
@@ -38,3 +40,13 @@ class CancelFlowRunTask(Task):
 
         client = Client()
         return client.cancel_flow_run(flow_run_id)
+
+
+class CancelFlowRunTask(CancelFlowRun):
+    def __new__(cls, *args, **kwargs):  # type: ignore
+        warnings.warn(
+            "`CancelFlowRunTask` has been renamed to `prefect.tasks.prefect.CancelFlowRun`,"
+            "please update your code accordingly",
+            stacklevel=2,
+        )
+        return super().__new__(cls)
diff --git a/src/prefect/tasks/prefect/flow_run_rename.py b/src/prefect/tasks/prefect/flow_run_rename.py
index 254bc3ae63..d4127a231c 100644
--- a/src/prefect/tasks/prefect/flow_run_rename.py
+++ b/src/prefect/tasks/prefect/flow_run_rename.py
@@ -1,3 +1,4 @@
+import warnings
 from typing import Any
 
 from prefect import Task
@@ -5,7 +6,7 @@ from prefect.client import Client
 from prefect.utilities.tasks import defaults_from_attrs
 
 
-class RenameFlowRunTask(Task):
+class RenameFlowRun(Task):
     """
     Task used to rename a running flow.
 
@@ -52,3 +53,13 @@ class RenameFlowRunTask(Task):
 
         client = Client()
         return client.set_flow_run_name(flow_run_id, flow_run_name)
+
+
+class RenameFlowRunTask(RenameFlowRun):
+    def __new__(cls, *args, **kwargs):  # type: ignore
+        warnings.warn(
+            "`RenameFlowRunTask` has been renamed to `prefect.tasks.prefect.RenameFlowRun`,"
+            "please update your code accordingly",
+            stacklevel=2,
+        )
+        return super().__new__(cls)


Test patch:
diff --git a/tests/tasks/prefect/test_flow_run.py b/tests/tasks/prefect/test_flow_run.py
index 011295ff27..1abcf688d6 100644
--- a/tests/tasks/prefect/test_flow_run.py
+++ b/tests/tasks/prefect/test_flow_run.py
@@ -2,7 +2,7 @@ import pytest
 from unittest.mock import MagicMock
 
 import prefect
-from prefect.tasks.prefect.flow_run import FlowRunTask
+from prefect.tasks.prefect.flow_run import StartFlowRun
 
 
 @pytest.fixture()
@@ -21,10 +21,20 @@ def client(monkeypatch):
     yield cloud_client
 
 
-class TestFlowRunTaskCloud:
+def test_deprecated_old_name():
+    from prefect.tasks.prefect import FlowRunTask
+
+    with pytest.warns(UserWarning, match="StartFlowRun"):
+        task = FlowRunTask(name="My flow run")
+
+    assert isinstance(task, StartFlowRun)
+    assert task.name == "My flow run"
+
+
+class TestStartFlowRunCloud:
     def test_initialization(self, cloud_api):
         # verify that the task is initialized as expected
-        task = FlowRunTask(
+        task = StartFlowRun(
             name="My Flow Run Task",
             checkpoint=False,
             project_name="Test Project",
@@ -43,7 +53,7 @@ class TestFlowRunTaskCloud:
 
     def test_flow_run_task(self, client, cloud_api):
         # verify that create_flow_run was called
-        task = FlowRunTask(
+        task = StartFlowRun(
             project_name="Test Project",
             flow_name="Test Flow",
             parameters={"test": "ing"},
@@ -67,7 +77,7 @@ class TestFlowRunTaskCloud:
 
     def test_flow_run_task_with_flow_run_id(self, client, cloud_api):
         # verify that create_flow_run was called
-        task = FlowRunTask(
+        task = StartFlowRun(
             project_name="Test Project",
             flow_name="Test Flow",
             parameters={"test": "ing"},
@@ -88,7 +98,7 @@ class TestFlowRunTaskCloud:
 
     def test_idempotency_key_uses_map_index_if_present(self, client, cloud_api):
         # verify that create_flow_run was called
-        task = FlowRunTask(project_name="Test Project", flow_name="Test Flow")
+        task = StartFlowRun(project_name="Test Project", flow_name="Test Flow")
 
         # verify that run returns the new flow run ID
         with prefect.context(flow_run_id="test-id", map_index=4):
@@ -105,28 +115,28 @@ class TestFlowRunTaskCloud:
 
     def test_flow_run_task_without_flow_name(self, cloud_api):
         # verify that a ValueError is raised without a flow name
-        task = FlowRunTask(project_name="Test Project")
+        task = StartFlowRun(project_name="Test Project")
         with pytest.raises(ValueError, match="Must provide a flow name."):
             task.run()
 
     def test_flow_run_task_without_project_name(self, cloud_api):
         # verify that a ValueError is raised without a project name
-        task = FlowRunTask(flow_name="Test Flow")
+        task = StartFlowRun(flow_name="Test Flow")
         with pytest.raises(ValueError, match="Must provide a project name."):
             task.run()
 
     def test_flow_run_task_with_no_matching_flow(self, client, cloud_api):
         # verify a ValueError is raised if the client returns no flows
-        task = FlowRunTask(flow_name="Test Flow", project_name="Test Project")
+        task = StartFlowRun(flow_name="Test Flow", project_name="Test Project")
         client.graphql = MagicMock(return_value=MagicMock(data=MagicMock(flow=[])))
         with pytest.raises(ValueError, match="Flow 'Test Flow' not found."):
             task.run()
 
 
-class TestFlowRunTaskCoreServer:
+class TestStartFlowRunServer:
     def test_initialization(self, server_api):
         # verify that the task is initialized as expected
-        task = FlowRunTask(
+        task = StartFlowRun(
             name="My Flow Run Task",
             project_name="Demo",
             checkpoint=False,
@@ -144,7 +154,7 @@ class TestFlowRunTaskCoreServer:
 
     def test_flow_run_task(self, client, server_api):
         # verify that create_flow_run was called
-        task = FlowRunTask(
+        task = StartFlowRun(
             flow_name="Test Flow",
             project_name="Demo",
             parameters={"test": "ing"},
@@ -167,20 +177,20 @@ class TestFlowRunTaskCoreServer:
 
     def test_flow_run_task_without_flow_name(self, server_api):
         # verify that a ValueError is raised without a flow name
-        task = FlowRunTask()
+        task = StartFlowRun()
         with pytest.raises(ValueError, match="Must provide a flow name."):
             task.run()
 
     def test_flow_run_task_with_no_matching_flow(self, client, server_api):
         # verify a ValueError is raised if the client returns no flows
-        task = FlowRunTask(flow_name="Test Flow", project_name="Demo")
+        task = StartFlowRun(flow_name="Test Flow", project_name="Demo")
         client.graphql = MagicMock(return_value=MagicMock(data=MagicMock(flow=[])))
         with pytest.raises(ValueError, match="Flow 'Test Flow' not found."):
             task.run()
 
     def test_flow_run_task_with_flow_run_id(self, client, server_api):
         # verify that create_flow_run was called
-        task = FlowRunTask(
+        task = StartFlowRun(
             project_name="Test Project",
             flow_name="Test Flow",
             parameters={"test": "ing"},
diff --git a/tests/tasks/prefect/test_flow_run_cancel.py b/tests/tasks/prefect/test_flow_run_cancel.py
index 9d2f371693..f2077f6d0f 100644
--- a/tests/tasks/prefect/test_flow_run_cancel.py
+++ b/tests/tasks/prefect/test_flow_run_cancel.py
@@ -1,9 +1,18 @@
 import pytest
 from unittest.mock import MagicMock
 
-import prefect
 from prefect import Flow
-from prefect.tasks.prefect.flow_run_cancel import CancelFlowRunTask
+from prefect.tasks.prefect.flow_run_cancel import CancelFlowRun
+
+
+def test_deprecated_old_name():
+    from prefect.tasks.prefect import CancelFlowRunTask
+
+    with pytest.warns(UserWarning, match="`prefect.tasks.prefect.CancelFlowRun`"):
+        task = CancelFlowRunTask(flow_run_id="id123")
+
+    assert isinstance(task, CancelFlowRun)
+    assert task.flow_run_id == "id123"
 
 
 def test_flow_run_cancel(monkeypatch):
@@ -12,7 +21,7 @@ def test_flow_run_cancel(monkeypatch):
     monkeypatch.setattr(
         "prefect.tasks.prefect.flow_run_cancel.Client", MagicMock(return_value=client)
     )
-    flow_cancel_task = CancelFlowRunTask(flow_run_id="id123")
+    flow_cancel_task = CancelFlowRun(flow_run_id="id123")
 
     # Verify correct initialization
     assert flow_cancel_task.flow_run_id == "id123"
diff --git a/tests/tasks/prefect/test_flow_run_rename.py b/tests/tasks/prefect/test_flow_run_rename.py
index 3af58a812a..6801c0c7a2 100644
--- a/tests/tasks/prefect/test_flow_run_rename.py
+++ b/tests/tasks/prefect/test_flow_run_rename.py
@@ -1,8 +1,17 @@
 import pytest
 from unittest.mock import MagicMock
 
-import prefect
-from prefect.tasks.prefect.flow_run_rename import RenameFlowRunTask
+from prefect.tasks.prefect.flow_run_rename import RenameFlowRun
+
+
+def test_deprecated_old_name():
+    from prefect.tasks.prefect import RenameFlowRunTask
+
+    with pytest.warns(UserWarning, match="`prefect.tasks.prefect.RenameFlowRun`"):
+        task = RenameFlowRunTask(flow_run_id="id123")
+
+    assert isinstance(task, RenameFlowRun)
+    assert task.flow_run_id == "id123"
 
 
 def test_flow_run_rename_task(monkeypatch):
@@ -12,7 +21,7 @@ def test_flow_run_rename_task(monkeypatch):
         "prefect.tasks.prefect.flow_run_rename.Client", MagicMock(return_value=client)
     )
 
-    task = RenameFlowRunTask(flow_run_id="id123", flow_run_name="a_new_name!")
+    task = RenameFlowRun(flow_run_id="id123", flow_run_name="a_new_name!")
 
     # Verify correct initialization
     assert task.flow_run_id == "id123"
@@ -26,12 +35,12 @@ def test_flow_run_rename_task(monkeypatch):
 
 
 def test_missing_flow_run_id():
-    task = RenameFlowRunTask()
+    task = RenameFlowRun()
     with pytest.raises(ValueError, match="Must provide a flow run ID."):
         task.run(flow_run_name="a_new_name!")
 
 
 def test_missing_flow_run_name():
-    task = RenameFlowRunTask()
+    task = RenameFlowRun()
     with pytest.raises(ValueError, match="Must provide a flow name."):
         task.run(flow_run_id="id123")