Backport: deleting flows in python client prefect 2.0

src/prefect/client/orchestration.py

@@ -499,6 +499,23 @@ async def read_flow(self, flow_id: UUID) -> Flow:
         response = await self._client.get(f"/flows/{flow_id}")
         return Flow.parse_obj(response.json())
 
+    async def delete_flow(self, flow_id: UUID) -> None:
+        """
+        Delete a flow by UUID.
+        Args:
+            flow_id: ID of the flow to be deleted
+        Raises:
+            prefect.exceptions.ObjectNotFound: If request returns 404
+            httpx.RequestError: If requests fails
+        """
+        try:
+            await self._client.delete(f"/flows/{flow_id}")
+        except httpx.HTTPStatusError as e:
+            if e.response.status_code == status.HTTP_404_NOT_FOUND:
+                raise prefect.exceptions.ObjectNotFound(http_exc=e) from e
+            else:
+                raise
+
     async def read_flows(
         self,
         *,

tests/client/test_prefect_client.py

@@ -568,6 +568,19 @@ def foo():
     assert lookup.name == foo.name
 
 
+async def test_create_then_delete_flow(prefect_client):
+    @flow
+    def foo():
+        pass
+
+    flow_id = await prefect_client.create_flow(foo)
+    assert isinstance(flow_id, UUID)
+
+    await prefect_client.delete_flow(flow_id)
+    with pytest.raises(prefect.exceptions.PrefectHTTPStatusError, match="404"):
+        await prefect_client.read_flow(flow_id)
+
+
 async def test_create_then_read_deployment(
     prefect_client, infrastructure_document_id, storage_document_id
 ):