diff --git a/src/prefect/client/orchestration.py b/src/prefect/client/orchestration.py index dfbc3c119b6d..4a5a06790a65 100644 --- a/src/prefect/client/orchestration.py +++ b/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, *, diff --git a/tests/client/test_prefect_client.py b/tests/client/test_prefect_client.py index fce86c1a8040..b1da96987c2d 100644 --- a/tests/client/test_prefect_client.py +++ b/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 ):