Merge pull request #4021 from PrefectHQ/run_cli_id

Add option to provide flow ID when calling run flow CLI command

changes/pr4021.yaml

@@ -0,0 +1,3 @@
+enhancement:
+  - "Add option to provide flow ID to `run flow` CLI command - [#4021](https://github.com/PrefectHQ/prefect/pull/4021)"
+  - "Flow name and project are no longer required options when calling `run flow` CLI command - [#4021](https://github.com/PrefectHQ/prefect/pull/4021)"

src/prefect/cli/run.py

@@ -35,12 +35,15 @@ def run():
 
 @run.command(hidden=True)
 @click.option(
-    "--name", "-n", required=True, help="The name of a flow to run.", hidden=True
+    "--id", "-i", required=False, help="The id of a flow to run.", hidden=True
+)
+@click.option(
+    "--name", "-n", required=False, help="The name of a flow to run.", hidden=True
 )
 @click.option(
     "--project",
     "-p",
-    required=True,
+    required=False,
     help="The project that contains the flow.",
     hidden=True,
 )
@@ -74,6 +77,7 @@ def run():
     hidden=True,
 )
 def flow(
+    id,
     name,
     project,
     version,
@@ -90,9 +94,9 @@ def flow(
 
     \b
     Options:
-        --name, -n                  TEXT        The name of a flow to run [required]
-        --project, -p               TEXT        The name of a project that contains
-                                                the flow [required]
+        --id, -i                    TEXT        The ID of a flow to run
+        --name, -n                  TEXT        The name of a flow to run
+        --project, -p               TEXT        The name of a project that contains the flow
         --version, -v               INTEGER     A flow version to run
         --parameters-file, -pf      FILE PATH   A filepath of a JSON file containing
                                                 parameters
@@ -110,6 +114,9 @@ def flow(
         --no-url                                Only output the flow run id instead of a
                                                 link
 
+    \b
+    Either `id` or both `name` and `project` must be provided to run a flow.
+
     \b
     If both `--parameters-file` and `--parameters-string` are provided then the values
     passed in through the string will override the values provided from the file.
@@ -125,46 +132,62 @@ def flow(
         $ prefect run flow -n "Test-Flow" -p "My Project" -ps '{"my_param": 42}'
         Flow Run: https://cloud.prefect.io/myslug/flow-run/2ba3rrfd-411c-4d99-bb2a-f64a6dea78f9
     """
+    if not id and not (name and project):
+        click.secho(
+            "A flow ID or some combination of flow name and project must be provided.",
+            fg="red",
+        )
+        return
+
+    if id and (name or project):
+        click.secho(
+            "Both a flow ID and a name/project combination cannot be provided.",
+            fg="red",
+        )
+        return
+
     if watch and logs:
         click.secho(
             "Streaming state and logs not currently supported together.", fg="red"
         )
         return
 
-    where_clause = {
-        "_and": {
-            "name": {"_eq": name},
-            "version": {"_eq": version},
-            "project": {"name": {"_eq": project}},
+    client = Client()
+    flow_id = id
+    if not flow_id:
+        where_clause = {
+            "_and": {
+                "name": {"_eq": name},
+                "version": {"_eq": version},
+                "project": {"name": {"_eq": project}},
+            }
         }
-    }
 
-    query = {
-        "query": {
-            with_args(
-                "flow",
-                {
-                    "where": where_clause,
-                    "order_by": {
-                        "name": EnumValue("asc"),
-                        "version": EnumValue("desc"),
+        query = {
+            "query": {
+                with_args(
+                    "flow",
+                    {
+                        "where": where_clause,
+                        "order_by": {
+                            "name": EnumValue("asc"),
+                            "version": EnumValue("desc"),
+                        },
+                        "distinct_on": EnumValue("name"),
                     },
-                    "distinct_on": EnumValue("name"),
-                },
-            ): {"id": True}
+                ): {"id": True}
+            }
         }
-    }
 
-    client = Client()
-    result = client.graphql(query)
+        result = client.graphql(query)
 
-    flow_data = result.data.flow
+        flow_data = result.data.flow
 
-    if flow_data:
-        flow_id = flow_data[0].id
-    else:
-        click.secho("{} not found".format(name), fg="red")
-        return
+        if flow_data:
+            flow_id = flow_data[0].id
+        else:
+            click.secho("{} not found".format(name), fg="red")
+            return
 
     # Load parameters from file if provided
     file_params = {}

tests/cli/test_run.py

@@ -468,3 +468,67 @@ def test_run_flow_flow_run_id_no_link(monkeypatch, cloud_api):
     )
     assert result.exit_code == 0
     assert "Flow Run ID" in result.output
+
+
+def test_run_flow_using_id(monkeypatch, cloud_api):
+    post = MagicMock(
+        return_value=MagicMock(
+            json=MagicMock(return_value=dict(data=dict(flow=[{"id": "flow"}])))
+        )
+    )
+    session = MagicMock()
+    session.return_value.post = post
+    monkeypatch.setattr("requests.Session", session)
+
+    create_flow_run_mock = MagicMock(return_value="id")
+    monkeypatch.setattr("prefect.client.Client.create_flow_run", create_flow_run_mock)
+    monkeypatch.setattr(
+        "prefect.client.Client.get_default_tenant_slug", MagicMock(return_value="tslug")
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(
+        run,
+        [
+            "flow",
+            "--id",
+            "id",
+        ],
+    )
+    assert result.exit_code == 0
+    assert "Flow Run" in result.output
+    assert create_flow_run_mock.called
+
+
+def test_run_flow_no_id_or_name_and_project():
+    runner = CliRunner()
+    result = runner.invoke(
+        run,
+        [
+            "flow",
+        ],
+    )
+    assert (
+        "A flow ID or some combination of flow name and project must be provided"
+        in result.output
+    )
+
+
+def test_run_flow_no_id_or_name_and_project():
+    runner = CliRunner()
+    result = runner.invoke(
+        run,
+        [
+            "flow",
+            "--id",
+            "id",
+            "--name",
+            "flow",
+            "--project",
+            "project",
+        ],
+    )
+    assert (
+        "Both a flow ID and a name/project combination cannot be provided"
+        in result.output
+    )