`prefect deployment ls` errors if a deployment's associated flow has been deleted

[discdiver]

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

prefect deployment ls errors if a deployment's associated flow has been deleted. Ideally that case would be handled gracefully.

Reproduction

Create a deployment.

Delete the associated flow.

Error

Traceback (most recent call last):
  File "/Users/jeffhale/miniforge3/lib/python3.10/site-packages/prefect/cli/_utilities.py", line 41, in wrapper
    return fn(*args, **kwargs)
  File "/Users/jeffhale/miniforge3/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 255, in coroutine_wrapper
    return call()
  File "/Users/jeffhale/miniforge3/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 383, in __call__
    return self.result()
  File "/Users/jeffhale/miniforge3/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 283, in result
    return self.future.result(timeout=timeout)
  File "/Users/jeffhale/miniforge3/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 169, in result
    return self.__get_result()
  File "/Users/jeffhale/miniforge3/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/Users/jeffhale/miniforge3/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 346, in _run_async
    result = await coro
  File "/Users/jeffhale/miniforge3/lib/python3.10/site-packages/prefect/cli/deployment.py", line 460, in ls
    for deployment in sorted(
  File "/Users/jeffhale/miniforge3/lib/python3.10/site-packages/prefect/cli/deployment.py", line 449, in sort_by_name_keys
    return flows[d.flow_id].name, d.name
KeyError: UUID('ca4aa552-16d2-4dcd-bb0c-e3f8fac0483d')
An exception occurred.

### Versions

```Text
Version:             2.10.18
API version:         0.8.4
Python version:      3.10.8
Git commit:          cf177852
Built:               Thu, Jun 29, 2023 2:34 PM
OS/Arch:             darwin/arm64
Profile:             prod
Server type:         cloud

### Additional context

Diagnosis courtesy of @desertaxle. 

[serinamarie]

We should definitely raise a clearer error.

[serinamarie]

Actually, I think it would be ideal to update the logic here:

https://github.com/PrefectHQ/prefect/blob/f32b9c35e8ae8af5d74754663fb2d613a44c4d3d/src/prefect/cli/deployment.py#L437-L468

[alexmojaki]

I can reproduce this in prefect cloud, but not locally. Both in unit tests (as tried in https://github.com/PrefectHQ/prefect/pull/10122) and when running a server locally (with both Postgres and SQLite) I find that deleting a flow automatically deletes the associated deployments.

This fits with the ORM model which says ondelete="CASCADE" meaning that the database should take care of the deletion automatically:

https://github.com/PrefectHQ/prefect/blob/7296eedaacdf69217fa153e7619d311a6ad2a179/src/prefect/server/database/orm_models.py#L850-L856

So for some reason it seems prefect cloud is behaving differently?

[serinamarie]

@alexmojaki, thanks for pointing this out. I think you may be right. I'll take a look soon.

[serinamarie]

Ah yeah, pretty sure it's due to the fact that in oss we actually delete the flow on the spot rather than simply soft delete it.

oss: https://github.com/PrefectHQ/prefect/blob/1998adfdcc8760289b46235714d3f6bbad9fa8cd/src/prefect/server/api/flows.py#L150-L162

However, the solution I posed in #10122 seems to fix the issue when using a Cloud profile as well.

I think the trick will be mocking that soft deletion of the flow for the test. cc @rito-sixt which I imagine when you make the call to retrieve all available flows, intercept that call and return the list of flows with the "soft deleted" flow subtracted.

[jsopkin]

Just curious here- is the expected behavior that deleting a flow should not delete associated deployments?

[brodul]

Has someone figured out a workaround?

[brodul]

For folks with a deadlines, I created a patch that does not solve the underling issue, but should be fine for CICD pipelines. This is a hack, don't use if you don't know what you are doing

Copy contents to patchfile

diff --git a/src/prefect/cli/deployment.py b/src/prefect/cli/deployment.py
index 38b3da40f..3e26abb50 100644
--- a/src/prefect/cli/deployment.py
+++ b/src/prefect/cli/deployment.py
@@ -1,6 +1,7 @@
 """
 Command line interface for working with deployments.
 """
+
 import json
 import sys
 import textwrap
@@ -461,7 +462,7 @@ async def ls(flow_name: List[str] = None, by_created: bool = False):
         }
 
     def sort_by_name_keys(d):
-        return flows[d.flow_id].name, d.name
+        return getattr(flows.get(d.flow_id), "name", ""), d.name
I DON'T KNOW WHAT I AM DOING
 
     def sort_by_created_key(d):
         return pendulum.now("utc") - d.created
@@ -475,10 +476,11 @@ async def ls(flow_name: List[str] = None, by_created: bool = False):
     for deployment in sorted(
         deployments, key=sort_by_created_key if by_created else sort_by_name_keys
     ):
-        table.add_row(
-            f"{flows[deployment.flow_id].name}/[bold]{deployment.name}[/]",
-            str(deployment.id),
-        )
+        if deployment.flow_id in flows:
+            table.add_row(
+                f"{flows[deployment.flow_id].name}/[bold]{deployment.name}[/]",
+                str(deployment.id),
+            )
 
     app.console.print(table)

patch -p1 /utils/venv/lib/python3.11/site-packages/prefect/cli/deployment.py < patchfile

This will apply on prefect==2.14.21

[serinamarie]

Just curious here- is the expected behavior that deleting a flow should not delete associated deployments?

@Bada-S, that's correct, flows are emergent objects.

I think I need to reopen the stale PR, and that solution should work.

[serinamarie]

@discdiver I'm no longer able to reproduce this in Cloud or OSS using Prefect version 2.18.0. Are you still able to? If not, I will close.

[brodul]

Error output: Traceback (most recent call last):
  File "/utils/venv/lib/python3.11/site-packages/prefect/cli/_utilities.py", line 42, in wrapper
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/utils/venv/lib/python3.11/site-packages/prefect/utilities/asyncutils.py", line 304, in coroutine_wrapper
    return call()
           ^^^^^^
  File "/utils/venv/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 432, in __call__
    return self.result()
           ^^^^^^^^^^^^^
  File "/utils/venv/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 318, in result
    return self.future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/utils/venv/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 179, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/utils/venv/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 389, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/utils/venv/lib/python3.11/site-packages/prefect/cli/deployment.py", line 822, in ls
    for deployment in sorted(
                      ^^^^^^^
  File "/utils/venv/lib/python3.11/site-packages/prefect/cli/deployment.py", line 811, in sort_by_name_keys
    return flows[d.flow_id].name, d.name
           ~~~~~^^^^^^^^^^^
KeyError: UUID('6XXX')

@serinamarie This is still a problem for us in Prefect Cloud. The prefect package version we are using is 2.19.4 .

[brodul]

I guess there is an issue with the data in the database.

Patch will apply on prefect==2.19.4

patch -p1 /utils/venv/lib/python3.11/site-packages/prefect/cli/deployment.py < patchfile

diff --git a/src/prefect/cli/deployment.py b/src/prefect/cli/deployment.py
index 8d642a736..e18c76a6d 100644
--- a/src/prefect/cli/deployment.py
+++ b/src/prefect/cli/deployment.py
@@ -808,7 +808,11 @@ async def ls(flow_name: Optional[List[str]] = None, by_created: bool = False):
         }

     def sort_by_name_keys(d):
-        return flows[d.flow_id].name, d.name
+        try:
+            return flows[d.flow_id].name, d.name
+        except KeyError:
+            app.console.print(f"BUG: Flow with flow id {d.flow_id} not found\n")
+            return "", d.name

     def sort_by_created_key(d):
         return pendulum.now("utc") - d.created
@@ -822,10 +826,15 @@ async def ls(flow_name: Optional[List[str]] = None, by_created: bool = False):
     for deployment in sorted(
         deployments, key=sort_by_created_key if by_created else sort_by_name_keys
     ):
-        table.add_row(
-            f"{flows[deployment.flow_id].name}/[bold]{deployment.name}[/]",
-            str(deployment.id),
-        )
+        if deployment.flow_id in flows:
+            table.add_row(
+                f"{flows[deployment.flow_id].name}/[bold]{deployment.name}[/]",
+                str(deployment.id),
+            )
+        else:
+            app.console.print(
+                f"BUG - Contact support: Flow with flow id {deployment.flow_id} not found, but is set in deployment with id {deployment.id}\n"
+            )

     app.console.print(table)

[brodul]

In the UI it looks something like this

[brodul]

In CLI after the patch