We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
no import statement for wait_for_flow_run
### First check

- [X] I added a descriptive title to this issue.
- [X] I used GitHub search to find a similar request and didn't find it 😇

### Describe the issue

There is no import statement for wait_for_flow_run so typing this code into pycharm shows wait_for_flow_run as an error.  Searching the internets, the import statement used to be

_from prefect.tasks.prefect import wait_for_flow_run_

yeah, that doesn't work anymore.

### Describe the proposed change

put the correct import statement in the docs which is 

_from prefect.flow_runs import wait_for_flow_run_


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/flow_runs.py`
Content:
```
1 from typing import Optional
2 from uuid import UUID
3 
4 import anyio
5 
6 from prefect.client.orchestration import PrefectClient
7 from prefect.client.schemas import FlowRun
8 from prefect.client.utilities import inject_client
9 from prefect.exceptions import FlowRunWaitTimeout
10 from prefect.logging import get_logger
11 
12 
13 @inject_client
14 async def wait_for_flow_run(
15     flow_run_id: UUID,
16     timeout: Optional[int] = 10800,
17     poll_interval: int = 5,
18     client: Optional[PrefectClient] = None,
19     log_states: bool = False,
20 ) -> FlowRun:
21     """
22     Waits for the prefect flow run to finish and returns the FlowRun
23 
24     Args:
25         flow_run_id: The flow run ID for the flow run to wait for.
26         timeout: The wait timeout in seconds. Defaults to 10800 (3 hours).
27         poll_interval: The poll interval in seconds. Defaults to 5.
28 
29     Returns:
30         FlowRun: The finished flow run.
31 
32     Raises:
33         prefect.exceptions.FlowWaitTimeout: If flow run goes over the timeout.
34 
35     Examples:
36         Create a flow run for a deployment and wait for it to finish:
37             ```python
38             import asyncio
39 
40             from prefect import get_client
41 
42             async def main():
43                 async with get_client() as client:
44                     flow_run = await client.create_flow_run_from_deployment(deployment_id="my-deployment-id")
45                     flow_run = await wait_for_flow_run(flow_run_id=flow_run.id)
46                     print(flow_run.state)
47 
48             if __name__ == "__main__":
49                 asyncio.run(main())
50 
51             ```
52 
53         Trigger multiple flow runs and wait for them to finish:
54             ```python
55             import asyncio
56 
57             from prefect import get_client
58 
59             async def main(num_runs: int):
60                 async with get_client() as client:
61                     flow_runs = [
62                         await client.create_flow_run_from_deployment(deployment_id="my-deployment-id")
63                         for _
64                         in range(num_runs)
65                     ]
66                     coros = [wait_for_flow_run(flow_run_id=flow_run.id) for flow_run in flow_runs]
67                     finished_flow_runs = await asyncio.gather(*coros)
68                     print([flow_run.state for flow_run in finished_flow_runs])
69 
70             if __name__ == "__main__":
71                 asyncio.run(main(num_runs=10))
72 
73             ```
74     """
75     assert client is not None, "Client injection failed"
76     logger = get_logger()
77     with anyio.move_on_after(timeout):
78         while True:
79             flow_run = await client.read_flow_run(flow_run_id)
80             flow_state = flow_run.state
81             if log_states:
82                 logger.info(f"Flow run is in state {flow_run.state.name!r}")
83             if flow_state and flow_state.is_final():
84                 return flow_run
85             await anyio.sleep(poll_interval)
86     raise FlowRunWaitTimeout(
87         f"Flow run with ID {flow_run_id} exceeded watch timeout of {timeout} seconds"
88     )
89
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/prefect/flow_runs.py b/src/prefect/flow_runs.py
--- a/src/prefect/flow_runs.py
+++ b/src/prefect/flow_runs.py
@@ -38,6 +38,7 @@
             import asyncio
 
             from prefect import get_client
+            from prefect.flow_runs import wait_for_flow_run
 
             async def main():
                 async with get_client() as client:
@@ -55,6 +56,7 @@
             import asyncio
 
             from prefect import get_client
+            from prefect.flow_runs import wait_for_flow_run
 
             async def main(num_runs: int):
                 async with get_client() as client:

{"golden_diff": "diff --git a/src/prefect/flow_runs.py b/src/prefect/flow_runs.py\n--- a/src/prefect/flow_runs.py\n+++ b/src/prefect/flow_runs.py\n@@ -38,6 +38,7 @@\n             import asyncio\n \n             from prefect import get_client\n+            from prefect.flow_runs import wait_for_flow_run\n \n             async def main():\n                 async with get_client() as client:\n@@ -55,6 +56,7 @@\n             import asyncio\n \n             from prefect import get_client\n+            from prefect.flow_runs import wait_for_flow_run\n \n             async def main(num_runs: int):\n                 async with get_client() as client:\n", "issue": "no import statement for wait_for_flow_run\n### First check\r\n\r\n- [X] I added a descriptive title to this issue.\r\n- [X] I used GitHub search to find a similar request and didn't find it \ud83d\ude07\r\n\r\n### Describe the issue\r\n\r\nThere is no import statement for wait_for_flow_run so typing this code into pycharm shows wait_for_flow_run as an error.  Searching the internets, the import statement used to be\r\n\r\n_from prefect.tasks.prefect import wait_for_flow_run_\r\n\r\nyeah, that doesn't work anymore.\r\n\r\n### Describe the proposed change\r\n\r\nput the correct import statement in the docs which is \r\n\r\n_from prefect.flow_runs import wait_for_flow_run_\r\n\n", "before_files": [{"content": "from typing import Optional\nfrom uuid import UUID\n\nimport anyio\n\nfrom prefect.client.orchestration import PrefectClient\nfrom prefect.client.schemas import FlowRun\nfrom prefect.client.utilities import inject_client\nfrom prefect.exceptions import FlowRunWaitTimeout\nfrom prefect.logging import get_logger\n\n\n@inject_client\nasync def wait_for_flow_run(\n    flow_run_id: UUID,\n    timeout: Optional[int] = 10800,\n    poll_interval: int = 5,\n    client: Optional[PrefectClient] = None,\n    log_states: bool = False,\n) -> FlowRun:\n    \"\"\"\n    Waits for the prefect flow run to finish and returns the FlowRun\n\n    Args:\n        flow_run_id: The flow run ID for the flow run to wait for.\n        timeout: The wait timeout in seconds. Defaults to 10800 (3 hours).\n        poll_interval: The poll interval in seconds. Defaults to 5.\n\n    Returns:\n        FlowRun: The finished flow run.\n\n    Raises:\n        prefect.exceptions.FlowWaitTimeout: If flow run goes over the timeout.\n\n    Examples:\n        Create a flow run for a deployment and wait for it to finish:\n            ```python\n            import asyncio\n\n            from prefect import get_client\n\n            async def main():\n                async with get_client() as client:\n                    flow_run = await client.create_flow_run_from_deployment(deployment_id=\"my-deployment-id\")\n                    flow_run = await wait_for_flow_run(flow_run_id=flow_run.id)\n                    print(flow_run.state)\n\n            if __name__ == \"__main__\":\n                asyncio.run(main())\n\n            ```\n\n        Trigger multiple flow runs and wait for them to finish:\n            ```python\n            import asyncio\n\n            from prefect import get_client\n\n            async def main(num_runs: int):\n                async with get_client() as client:\n                    flow_runs = [\n                        await client.create_flow_run_from_deployment(deployment_id=\"my-deployment-id\")\n                        for _\n                        in range(num_runs)\n                    ]\n                    coros = [wait_for_flow_run(flow_run_id=flow_run.id) for flow_run in flow_runs]\n                    finished_flow_runs = await asyncio.gather(*coros)\n                    print([flow_run.state for flow_run in finished_flow_runs])\n\n            if __name__ == \"__main__\":\n                asyncio.run(main(num_runs=10))\n\n            ```\n    \"\"\"\n    assert client is not None, \"Client injection failed\"\n    logger = get_logger()\n    with anyio.move_on_after(timeout):\n        while True:\n            flow_run = await client.read_flow_run(flow_run_id)\n            flow_state = flow_run.state\n            if log_states:\n                logger.info(f\"Flow run is in state {flow_run.state.name!r}\")\n            if flow_state and flow_state.is_final():\n                return flow_run\n            await anyio.sleep(poll_interval)\n    raise FlowRunWaitTimeout(\n        f\"Flow run with ID {flow_run_id} exceeded watch timeout of {timeout} seconds\"\n    )\n", "path": "src/prefect/flow_runs.py"}], "after_files": [{"content": "from typing import Optional\nfrom uuid import UUID\n\nimport anyio\n\nfrom prefect.client.orchestration import PrefectClient\nfrom prefect.client.schemas import FlowRun\nfrom prefect.client.utilities import inject_client\nfrom prefect.exceptions import FlowRunWaitTimeout\nfrom prefect.logging import get_logger\n\n\n@inject_client\nasync def wait_for_flow_run(\n    flow_run_id: UUID,\n    timeout: Optional[int] = 10800,\n    poll_interval: int = 5,\n    client: Optional[PrefectClient] = None,\n    log_states: bool = False,\n) -> FlowRun:\n    \"\"\"\n    Waits for the prefect flow run to finish and returns the FlowRun\n\n    Args:\n        flow_run_id: The flow run ID for the flow run to wait for.\n        timeout: The wait timeout in seconds. Defaults to 10800 (3 hours).\n        poll_interval: The poll interval in seconds. Defaults to 5.\n\n    Returns:\n        FlowRun: The finished flow run.\n\n    Raises:\n        prefect.exceptions.FlowWaitTimeout: If flow run goes over the timeout.\n\n    Examples:\n        Create a flow run for a deployment and wait for it to finish:\n            ```python\n            import asyncio\n\n            from prefect import get_client\n            from prefect.flow_runs import wait_for_flow_run\n\n            async def main():\n                async with get_client() as client:\n                    flow_run = await client.create_flow_run_from_deployment(deployment_id=\"my-deployment-id\")\n                    flow_run = await wait_for_flow_run(flow_run_id=flow_run.id)\n                    print(flow_run.state)\n\n            if __name__ == \"__main__\":\n                asyncio.run(main())\n\n            ```\n\n        Trigger multiple flow runs and wait for them to finish:\n            ```python\n            import asyncio\n\n            from prefect import get_client\n            from prefect.flow_runs import wait_for_flow_run\n\n            async def main(num_runs: int):\n                async with get_client() as client:\n                    flow_runs = [\n                        await client.create_flow_run_from_deployment(deployment_id=\"my-deployment-id\")\n                        for _\n                        in range(num_runs)\n                    ]\n                    coros = [wait_for_flow_run(flow_run_id=flow_run.id) for flow_run in flow_runs]\n                    finished_flow_runs = await asyncio.gather(*coros)\n                    print([flow_run.state for flow_run in finished_flow_runs])\n\n            if __name__ == \"__main__\":\n                asyncio.run(main(num_runs=10))\n\n            ```\n    \"\"\"\n    assert client is not None, \"Client injection failed\"\n    logger = get_logger()\n    with anyio.move_on_after(timeout):\n        while True:\n            flow_run = await client.read_flow_run(flow_run_id)\n            flow_state = flow_run.state\n            if log_states:\n                logger.info(f\"Flow run is in state {flow_run.state.name!r}\")\n            if flow_state and flow_state.is_final():\n                return flow_run\n            await anyio.sleep(poll_interval)\n    raise FlowRunWaitTimeout(\n        f\"Flow run with ID {flow_run_id} exceeded watch timeout of {timeout} seconds\"\n    )\n", "path": "src/prefect/flow_runs.py"}]}