prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <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_
  
  
  </issue>
  <code>
  [start of src/prefect/flow_runs.py]
  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 of src/prefect/flow_runs.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: 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:
  
verification_info: {"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"}]}