prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Filter by work pool not filtering the "dot" graph
  ### First check
  
  - [X] I added a descriptive title to this issue.
  - [X] I used the GitHub search to find a similar issue and didn't find it.
  - [X] I searched the Prefect documentation for this issue.
  - [X] I checked that this issue is related to Prefect and not one of its dependencies.
  
  ### Bug summary
  
  When filtering by "work pool" in the flows screen, the "dots graph" is not filtered.
  The "main-pool" is a pool I've just created, and there are no runs associated to it.
  
  
  ![CleanShot 2023-03-23 at 08 02 45@2x](https://user-images.githubusercontent.com/5398977/227117523-885b8337-37cc-4c87-80a4-f58b8c5e2526.jpg)
  
  
  ### Reproduction
  
  ```python3
  Create a new "work flow".
  Make sure it's empty and has no runs in it.
  Go to the flows screen, and filter by this work pool.
  You'll see that all the dots on the graph remain although they are not related to that work pool.
  ```
  
  
  ### Error
  
  _No response_
  
  ### Versions
  
  ```Text
  Prefect cloud
  ```
  
  
  ### Additional context
  
  _No response_
  
  --- 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/server/api/ui/flow_runs.py`
  Content:
  ```
  1 import datetime
  2 from typing import List
  3 from uuid import UUID
  4 
  5 from fastapi import Body, Depends
  6 from pydantic import Field
  7 
  8 import prefect.server.schemas as schemas
  9 from prefect.logging import get_logger
  10 from prefect.server import models
  11 from prefect.server.database.dependencies import provide_database_interface
  12 from prefect.server.database.interface import PrefectDBInterface
  13 from prefect.server.utilities.schemas import DateTimeTZ, PrefectBaseModel
  14 from prefect.server.utilities.server import PrefectRouter
  15 
  16 logger = get_logger("server.api.ui.flow_runs")
  17 
  18 router = PrefectRouter(prefix="/ui/flow_runs", tags=["Flow Runs", "UI"])
  19 
  20 
  21 class SimpleFlowRun(PrefectBaseModel):
  22     id: UUID = Field(default=..., description="The flow run id.")
  23     state_type: schemas.states.StateType = Field(
  24         default=..., description="The state type."
  25     )
  26     timestamp: DateTimeTZ = Field(
  27         default=...,
  28         description=(
  29             "The start time of the run, or the expected start time "
  30             "if it hasn't run yet."
  31         ),
  32     )
  33     duration: datetime.timedelta = Field(
  34         default=..., description="The total run time of the run."
  35     )
  36     lateness: datetime.timedelta = Field(
  37         default=..., description="The delay between the expected and actual start time."
  38     )
  39 
  40 
  41 @router.post("/history")
  42 async def read_flow_run_history(
  43     sort: schemas.sorting.FlowRunSort = Body(
  44         schemas.sorting.FlowRunSort.EXPECTED_START_TIME_DESC
  45     ),
  46     limit: int = Body(1000, le=1000),
  47     offset: int = Body(0, ge=0),
  48     flows: schemas.filters.FlowFilter = None,
  49     flow_runs: schemas.filters.FlowRunFilter = None,
  50     task_runs: schemas.filters.TaskRunFilter = None,
  51     deployments: schemas.filters.DeploymentFilter = None,
  52     db: PrefectDBInterface = Depends(provide_database_interface),
  53 ) -> List[SimpleFlowRun]:
  54     columns = [
  55         db.FlowRun.id,
  56         db.FlowRun.state_type,
  57         db.FlowRun.start_time,
  58         db.FlowRun.expected_start_time,
  59         db.FlowRun.total_run_time,
  60         # Although it isn't returned, we need to select
  61         # this field in order to compute `estimated_run_time`
  62         db.FlowRun.state_timestamp,
  63     ]
  64     async with db.session_context() as session:
  65         result = await models.flow_runs.read_flow_runs(
  66             columns=columns,
  67             flow_filter=flows,
  68             flow_run_filter=flow_runs,
  69             task_run_filter=task_runs,
  70             deployment_filter=deployments,
  71             sort=sort,
  72             limit=limit,
  73             offset=offset,
  74             session=session,
  75         )
  76     return [
  77         SimpleFlowRun(
  78             id=r.id,
  79             state_type=r.state_type,
  80             timestamp=r.start_time or r.expected_start_time,
  81             duration=r.estimated_run_time,
  82             lateness=r.estimated_start_time_delta,
  83         )
  84         for r in result
  85     ]
  86
  ```
  --- 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()
  
  ```
  
golden_diff: diff --git a/src/prefect/server/api/ui/flow_runs.py b/src/prefect/server/api/ui/flow_runs.py
  --- a/src/prefect/server/api/ui/flow_runs.py
  +++ b/src/prefect/server/api/ui/flow_runs.py
  @@ -49,6 +49,7 @@
       flow_runs: schemas.filters.FlowRunFilter = None,
       task_runs: schemas.filters.TaskRunFilter = None,
       deployments: schemas.filters.DeploymentFilter = None,
  +    work_pools: schemas.filters.WorkPoolFilter = None,
       db: PrefectDBInterface = Depends(provide_database_interface),
   ) -> List[SimpleFlowRun]:
       columns = [
  @@ -68,6 +69,7 @@
               flow_run_filter=flow_runs,
               task_run_filter=task_runs,
               deployment_filter=deployments,
  +            work_pool_filter=work_pools,
               sort=sort,
               limit=limit,
               offset=offset,
  
verification_info: {"golden_diff": "diff --git a/src/prefect/server/api/ui/flow_runs.py b/src/prefect/server/api/ui/flow_runs.py\n--- a/src/prefect/server/api/ui/flow_runs.py\n+++ b/src/prefect/server/api/ui/flow_runs.py\n@@ -49,6 +49,7 @@\n     flow_runs: schemas.filters.FlowRunFilter = None,\n     task_runs: schemas.filters.TaskRunFilter = None,\n     deployments: schemas.filters.DeploymentFilter = None,\n+    work_pools: schemas.filters.WorkPoolFilter = None,\n     db: PrefectDBInterface = Depends(provide_database_interface),\n ) -> List[SimpleFlowRun]:\n     columns = [\n@@ -68,6 +69,7 @@\n             flow_run_filter=flow_runs,\n             task_run_filter=task_runs,\n             deployment_filter=deployments,\n+            work_pool_filter=work_pools,\n             sort=sort,\n             limit=limit,\n             offset=offset,\n", "issue": "Filter by work pool not filtering the \"dot\" graph\n### First check\n\n- [X] I added a descriptive title to this issue.\n- [X] I used the GitHub search to find a similar issue and didn't find it.\n- [X] I searched the Prefect documentation for this issue.\n- [X] I checked that this issue is related to Prefect and not one of its dependencies.\n\n### Bug summary\n\nWhen filtering by \"work pool\" in the flows screen, the \"dots graph\" is not filtered.\r\nThe \"main-pool\" is a pool I've just created, and there are no runs associated to it.\r\n\r\n\r\n![CleanShot 2023-03-23 at 08 02 45@2x](https://user-images.githubusercontent.com/5398977/227117523-885b8337-37cc-4c87-80a4-f58b8c5e2526.jpg)\r\n\n\n### Reproduction\n\n```python3\nCreate a new \"work flow\".\r\nMake sure it's empty and has no runs in it.\r\nGo to the flows screen, and filter by this work pool.\r\nYou'll see that all the dots on the graph remain although they are not related to that work pool.\n```\n\n\n### Error\n\n_No response_\n\n### Versions\n\n```Text\nPrefect cloud\n```\n\n\n### Additional context\n\n_No response_\n", "before_files": [{"content": "import datetime\nfrom typing import List\nfrom uuid import UUID\n\nfrom fastapi import Body, Depends\nfrom pydantic import Field\n\nimport prefect.server.schemas as schemas\nfrom prefect.logging import get_logger\nfrom prefect.server import models\nfrom prefect.server.database.dependencies import provide_database_interface\nfrom prefect.server.database.interface import PrefectDBInterface\nfrom prefect.server.utilities.schemas import DateTimeTZ, PrefectBaseModel\nfrom prefect.server.utilities.server import PrefectRouter\n\nlogger = get_logger(\"server.api.ui.flow_runs\")\n\nrouter = PrefectRouter(prefix=\"/ui/flow_runs\", tags=[\"Flow Runs\", \"UI\"])\n\n\nclass SimpleFlowRun(PrefectBaseModel):\n    id: UUID = Field(default=..., description=\"The flow run id.\")\n    state_type: schemas.states.StateType = Field(\n        default=..., description=\"The state type.\"\n    )\n    timestamp: DateTimeTZ = Field(\n        default=...,\n        description=(\n            \"The start time of the run, or the expected start time \"\n            \"if it hasn't run yet.\"\n        ),\n    )\n    duration: datetime.timedelta = Field(\n        default=..., description=\"The total run time of the run.\"\n    )\n    lateness: datetime.timedelta = Field(\n        default=..., description=\"The delay between the expected and actual start time.\"\n    )\n\n\n@router.post(\"/history\")\nasync def read_flow_run_history(\n    sort: schemas.sorting.FlowRunSort = Body(\n        schemas.sorting.FlowRunSort.EXPECTED_START_TIME_DESC\n    ),\n    limit: int = Body(1000, le=1000),\n    offset: int = Body(0, ge=0),\n    flows: schemas.filters.FlowFilter = None,\n    flow_runs: schemas.filters.FlowRunFilter = None,\n    task_runs: schemas.filters.TaskRunFilter = None,\n    deployments: schemas.filters.DeploymentFilter = None,\n    db: PrefectDBInterface = Depends(provide_database_interface),\n) -> List[SimpleFlowRun]:\n    columns = [\n        db.FlowRun.id,\n        db.FlowRun.state_type,\n        db.FlowRun.start_time,\n        db.FlowRun.expected_start_time,\n        db.FlowRun.total_run_time,\n        # Although it isn't returned, we need to select\n        # this field in order to compute `estimated_run_time`\n        db.FlowRun.state_timestamp,\n    ]\n    async with db.session_context() as session:\n        result = await models.flow_runs.read_flow_runs(\n            columns=columns,\n            flow_filter=flows,\n            flow_run_filter=flow_runs,\n            task_run_filter=task_runs,\n            deployment_filter=deployments,\n            sort=sort,\n            limit=limit,\n            offset=offset,\n            session=session,\n        )\n    return [\n        SimpleFlowRun(\n            id=r.id,\n            state_type=r.state_type,\n            timestamp=r.start_time or r.expected_start_time,\n            duration=r.estimated_run_time,\n            lateness=r.estimated_start_time_delta,\n        )\n        for r in result\n    ]\n", "path": "src/prefect/server/api/ui/flow_runs.py"}], "after_files": [{"content": "import datetime\nfrom typing import List\nfrom uuid import UUID\n\nfrom fastapi import Body, Depends\nfrom pydantic import Field\n\nimport prefect.server.schemas as schemas\nfrom prefect.logging import get_logger\nfrom prefect.server import models\nfrom prefect.server.database.dependencies import provide_database_interface\nfrom prefect.server.database.interface import PrefectDBInterface\nfrom prefect.server.utilities.schemas import DateTimeTZ, PrefectBaseModel\nfrom prefect.server.utilities.server import PrefectRouter\n\nlogger = get_logger(\"server.api.ui.flow_runs\")\n\nrouter = PrefectRouter(prefix=\"/ui/flow_runs\", tags=[\"Flow Runs\", \"UI\"])\n\n\nclass SimpleFlowRun(PrefectBaseModel):\n    id: UUID = Field(default=..., description=\"The flow run id.\")\n    state_type: schemas.states.StateType = Field(\n        default=..., description=\"The state type.\"\n    )\n    timestamp: DateTimeTZ = Field(\n        default=...,\n        description=(\n            \"The start time of the run, or the expected start time \"\n            \"if it hasn't run yet.\"\n        ),\n    )\n    duration: datetime.timedelta = Field(\n        default=..., description=\"The total run time of the run.\"\n    )\n    lateness: datetime.timedelta = Field(\n        default=..., description=\"The delay between the expected and actual start time.\"\n    )\n\n\n@router.post(\"/history\")\nasync def read_flow_run_history(\n    sort: schemas.sorting.FlowRunSort = Body(\n        schemas.sorting.FlowRunSort.EXPECTED_START_TIME_DESC\n    ),\n    limit: int = Body(1000, le=1000),\n    offset: int = Body(0, ge=0),\n    flows: schemas.filters.FlowFilter = None,\n    flow_runs: schemas.filters.FlowRunFilter = None,\n    task_runs: schemas.filters.TaskRunFilter = None,\n    deployments: schemas.filters.DeploymentFilter = None,\n    work_pools: schemas.filters.WorkPoolFilter = None,\n    db: PrefectDBInterface = Depends(provide_database_interface),\n) -> List[SimpleFlowRun]:\n    columns = [\n        db.FlowRun.id,\n        db.FlowRun.state_type,\n        db.FlowRun.start_time,\n        db.FlowRun.expected_start_time,\n        db.FlowRun.total_run_time,\n        # Although it isn't returned, we need to select\n        # this field in order to compute `estimated_run_time`\n        db.FlowRun.state_timestamp,\n    ]\n    async with db.session_context() as session:\n        result = await models.flow_runs.read_flow_runs(\n            columns=columns,\n            flow_filter=flows,\n            flow_run_filter=flow_runs,\n            task_run_filter=task_runs,\n            deployment_filter=deployments,\n            work_pool_filter=work_pools,\n            sort=sort,\n            limit=limit,\n            offset=offset,\n            session=session,\n        )\n    return [\n        SimpleFlowRun(\n            id=r.id,\n            state_type=r.state_type,\n            timestamp=r.start_time or r.expected_start_time,\n            duration=r.estimated_run_time,\n            lateness=r.estimated_start_time_delta,\n        )\n        for r in result\n    ]\n", "path": "src/prefect/server/api/ui/flow_runs.py"}]}