Repository: PrefectHQ/prefect

Issue:
Chain `set_upstream` with Task Results
## Current behavior
<!-- Please describe how the feature works today -->
The `set_upstream` method is available on the functional API for tasks to explicitly set upstream dependencies like so:
```
from prefect import task, Flow


@task
def foo():
    return "foo"


@task
def bar():
    return "bar"


@task
def oops(x):
    return "oops"


with Flow("test") as flow:
    bar_result = bar()
    foo_result = foo()
    foo_result.set_upstream(bar_result)
    oops_result = oops(foo_result)
```



## Proposed behavior
<!-- Please describe your proposed change to the current behavior -->
It would be intuitive to use the API with chaining, like so:

```
from prefect import task, Flow


@task
def foo():
    return "foo"


@task
def bar():
    return "bar"


@task
def oops(x):
    return "oops"


with Flow("test") as flow:
    bar_result = bar()
    foo_result = foo().set_upstream(bar_result)
    oops_results = oops(foo_result)
```


Discussion:

Patch:
diff --git a/changes/issue3695.yaml b/changes/issue3695.yaml
new file mode 100644
index 0000000000..86b5295ae3
--- /dev/null
+++ b/changes/issue3695.yaml
@@ -0,0 +1,3 @@
+enhancement:
+  - "Allow chaining of `Task` imperative dependency calls - [#3696](https://github.com/PrefectHQ/prefect/pull/3696)"
+
diff --git a/changes/pr3692.yaml b/changes/pr3692.yaml
deleted file mode 100644
index 66c7042cc7..0000000000
--- a/changes/pr3692.yaml
+++ /dev/null
@@ -1,5 +0,0 @@
-task:
-  - "Add enhancement to `StartFlowRun` task to create link artifact for started flow run - [#3692](https://github.com/PrefectHQ/prefect/pull/3692)"
-
-enhancement:
-  - "Always use tenant slug in output of Client `get_cloud_url` function - [#3692](https://github.com/PrefectHQ/prefect/pull/3692)"
diff --git a/src/prefect/client/client.py b/src/prefect/client/client.py
index b10df0706a..aa04ff14b8 100644
--- a/src/prefect/client/client.py
+++ b/src/prefect/client/client.py
@@ -870,23 +870,25 @@ class Client:
         # returns "https://cloud.prefect.io/my-tenant-slug/flow-run/424242-ca-94611-111-55"
         ```
         """
+        # Generate direct link to UI
+        if prefect.config.backend == "cloud":
+            tenant_slug = self.get_default_tenant_slug(as_user=as_user)
+        else:
+            tenant_slug = ""
 
-        # Search for matching cloud API because we can't guarantee that the backend config is set
-        using_cloud_api = ".prefect.io" in prefect.config.cloud.api
-        tenant_slug = self.get_default_tenant_slug(as_user=as_user and using_cloud_api)
-
-        # For various API versions parse out `api-` for direct UI link
         base_url = (
-            (
-                re.sub("api-", "", prefect.config.cloud.api)
-                if re.search("api-", prefect.config.cloud.api)
-                else re.sub("api", "cloud", prefect.config.cloud.api)
-            )
-            if using_cloud_api
-            else prefect.config.server.ui.endpoint
+            re.sub("api-", "", prefect.config.cloud.api)
+            if re.search("api-", prefect.config.cloud.api)
+            else re.sub("api", "cloud", prefect.config.cloud.api)
         )
 
-        return "/".join([base_url.rstrip("/"), tenant_slug, subdirectory, id])
+        full_url = prefect.config.cloud.api
+        if tenant_slug:
+            full_url = "/".join([base_url.rstrip("/"), tenant_slug, subdirectory, id])
+        elif prefect.config.backend == "server":
+            full_url = "/".join([prefect.config.server.ui.endpoint, subdirectory, id])
+
+        return full_url
 
     def get_default_tenant_slug(self, as_user: bool = True) -> str:
         """
diff --git a/src/prefect/core/task.py b/src/prefect/core/task.py
index a0dcf77354..636cb4957a 100644
--- a/src/prefect/core/task.py
+++ b/src/prefect/core/task.py
@@ -635,7 +635,7 @@ class Task(metaclass=SignatureValidator):
         keyword_tasks: Mapping[str, object] = None,
         mapped: bool = False,
         validate: bool = None,
-    ) -> None:
+    ) -> "Task":
         """
         Set dependencies for a flow either specified or in the current context using this task
 
@@ -653,7 +653,7 @@ class Task(metaclass=SignatureValidator):
                 configuration file.
 
         Returns:
-            - None
+            - self
 
         Raises:
             - ValueError: if no flow is specified and no flow can be found in the current context
@@ -673,9 +673,11 @@ class Task(metaclass=SignatureValidator):
             mapped=mapped,
         )
 
+        return self
+
     def set_upstream(
         self, task: object, flow: "Flow" = None, key: str = None, mapped: bool = False
-    ) -> None:
+    ) -> "Task":
         """
         Sets the provided task as an upstream dependency of this task.
 
@@ -690,7 +692,7 @@ class Task(metaclass=SignatureValidator):
             - mapped (bool, optional): Whether this dependency is mapped; defaults to `False`
 
         Returns:
-            - None
+            - self
 
         Raises:
             - ValueError: if no flow is specified and no flow can be found in the current context
@@ -701,9 +703,11 @@ class Task(metaclass=SignatureValidator):
         else:
             self.set_dependencies(flow=flow, upstream_tasks=[task], mapped=mapped)
 
+        return self
+
     def set_downstream(
         self, task: "Task", flow: "Flow" = None, key: str = None, mapped: bool = False
-    ) -> None:
+    ) -> "Task":
         """
         Sets the provided task as a downstream dependency of this task.
 
@@ -715,6 +719,9 @@ class Task(metaclass=SignatureValidator):
                 will be passed to the downstream task's `run()` method under this keyword argument.
             - mapped (bool, optional): Whether this dependency is mapped; defaults to `False`
 
+        Returns:
+            - self
+
         Raises:
             - ValueError: if no flow is specified and no flow can be found in the current context
         """
@@ -726,6 +733,8 @@ class Task(metaclass=SignatureValidator):
         else:
             task.set_dependencies(flow=flow, upstream_tasks=[self], mapped=mapped)
 
+        return self
+
     def inputs(self) -> Dict[str, Dict]:
         """
         Describe the inputs for this task. The result is a dictionary that maps each input to
diff --git a/src/prefect/tasks/prefect/flow_run.py b/src/prefect/tasks/prefect/flow_run.py
index e5fe15fdb1..ada86d1cbb 100644
--- a/src/prefect/tasks/prefect/flow_run.py
+++ b/src/prefect/tasks/prefect/flow_run.py
@@ -2,10 +2,8 @@ import time
 import datetime
 import warnings
 from typing import Any
-from urllib.parse import urlparse
 
 from prefect import context, Task
-from prefect.artifacts import create_link
 from prefect.client import Client
 from prefect.engine.signals import signal_from_state
 from prefect.utilities.graphql import EnumValue, with_args
@@ -31,7 +29,7 @@ class StartFlowRun(Task):
             task will wait until the flow run is complete, and then reflect the corresponding
             state as the state of this task.  Defaults to `False`.
         - scheduled_start_time (datetime, optional): the time to schedule the execution
-            for; if not provided, defaults to now
+                for; if not provided, defaults to now
         - **kwargs (dict, optional): additional keyword arguments to pass to the Task constructor
     """
 
@@ -168,10 +166,6 @@ class StartFlowRun(Task):
 
         self.logger.debug(f"Flow Run {flow_run_id} created.")
 
-        self.logger.debug(f"Creating link artifact for Flow Run {flow_run_id}.")
-        run_link = client.get_cloud_url("flow-run", flow_run_id)
-        create_link(urlparse(run_link).path)
-
         if not self.wait:
             return flow_run_id
 


Test patch:
diff --git a/tests/core/test_task.py b/tests/core/test_task.py
index 43889c8bc8..8339efe247 100644
--- a/tests/core/test_task.py
+++ b/tests/core/test_task.py
@@ -529,6 +529,21 @@ class TestDependencies:
             t2.set_upstream(t1, **props)
             assert Edge(t1, t2, **props) in f.edges
 
+    def test_set_dependencies_stream_allows_chaining(self):
+        t1 = Task()
+        t2 = Task()
+        t3 = Task()
+        with Flow(name="test") as f:
+            t1_result = t1()
+            t2_result = t2()
+            t3_result = t3()
+
+            assert t1_result.set_downstream(t2_result) is t1_result
+            assert t3_result.set_upstream(t2_result) is t3_result
+            assert (
+                t3_result.set_dependencies(f, upstream_tasks=[t1_result]) is t3_result
+            )
+
 
 class TestSerialization:
     def test_serialization(self):
diff --git a/tests/tasks/prefect/test_flow_run.py b/tests/tasks/prefect/test_flow_run.py
index 137907cc05..ddca4a108a 100644
--- a/tests/tasks/prefect/test_flow_run.py
+++ b/tests/tasks/prefect/test_flow_run.py
@@ -15,15 +15,10 @@ def client(monkeypatch):
             )
         ),
         create_flow_run=MagicMock(return_value="xyz890"),
-        get_cloud_url=MagicMock(return_value="https://api.prefect.io/flow/run/url"),
-        create_task_run_artifact=MagicMock(return_value="id"),
     )
     monkeypatch.setattr(
         "prefect.tasks.prefect.flow_run.Client", MagicMock(return_value=cloud_client)
     )
-    monkeypatch.setattr(
-        "prefect.artifacts.Client", MagicMock(return_value=cloud_client)
-    )
     yield cloud_client
 
 
@@ -166,20 +161,6 @@ class TestStartFlowRunCloud:
         with pytest.raises(ValueError, match="Flow 'Test Flow' not found."):
             task.run()
 
-    def test_flow_run_link_artifact(self, client, cloud_api):
-        task = StartFlowRun(
-            project_name="Test Project",
-            flow_name="Test Flow",
-            parameters={"test": "ing"},
-            run_name="test-run",
-        )
-        with prefect.context(running_with_backend=True, task_run_id="trid"):
-            task.run()
-
-            client.create_task_run_artifact.assert_called_once_with(
-                data={"link": "/flow/run/url"}, kind="link", task_run_id="trid"
-            )
-
 
 class TestStartFlowRunServer:
     def test_initialization(self, server_api):