[PFS] support PFS delete run (#1859)

# Description

support PFS delete

# All Promptflow Contribution checklist:
- [x] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [ ] **I have read the [contribution guidelines](../CONTRIBUTING.md).**
- [ ] **Create an issue and link to the pull request to get dedicated
review from promptflow team. Learn more: [suggested
workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [ ] Title of the pull request is clear and informative.
- [ ] There are a small number of commits, each of which have an
informative message. This means that previously merged commits do not
appear in the history of the PR. For more information on cleaning up the
commits in your PR, [see this
page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [ ] Pull request includes test coverage for the included changes.

src/promptflow/promptflow/_sdk/_service/apis/connection.py

@@ -10,7 +10,7 @@
 import promptflow._sdk.schemas._connection as connection
 from promptflow._sdk._configuration import Configuration
 from promptflow._sdk._service import Namespace, Resource, fields
-from promptflow._sdk._service.utils.utils import build_pfs_user_agent, local_user_only
+from promptflow._sdk._service.utils.utils import build_pfs_user_agent, local_user_only, make_response_no_content
 from promptflow._sdk.entities._connection import _Connection
 
 api = Namespace("Connections", description="Connections Management")
@@ -146,12 +146,14 @@ def put(self, name: str):
 
     @api.doc(description="Delete connection")
     @local_user_only
+    @api.response(code=204, description="Delete connection", model=dict_field)
     @api.response(
         code=403, description="This service is available for local user only, please specify X-Remote-User in headers."
     )
     def delete(self, name: str):
         connection_op = _get_connection_operation()
         connection_op.delete(name=name)
+        return make_response_no_content()
 
 
 @api.route("/<string:name>/listsecrets")

src/promptflow/promptflow/_sdk/_service/apis/run.py

@@ -14,7 +14,7 @@
 from promptflow._sdk._constants import FlowRunProperties, get_list_view_type
 from promptflow._sdk._errors import RunNotFoundError
 from promptflow._sdk._service import Namespace, Resource, fields
-from promptflow._sdk._service.utils.utils import build_pfs_user_agent, get_client_from_request
+from promptflow._sdk._service.utils.utils import build_pfs_user_agent, get_client_from_request, make_response_no_content
 from promptflow._sdk.entities import Run as RunEntity
 from promptflow._sdk.operations._local_storage_operations import LocalStorageOperations
 from promptflow._utils.yaml_utils import dump_yaml
@@ -118,6 +118,12 @@ def get(self, name: str):
         run = get_client_from_request().runs.get(name=name)
         return jsonify(run._to_dict())
 
+    @api.response(code=204, description="Delete run", model=dict_field)
+    @api.doc(description="Delete run")
+    def delete(self, name: str):
+        get_client_from_request().runs.delete(name=name)
+        return make_response_no_content()
+
 
 @api.route("/<string:name>/childRuns")
 class FlowChildRuns(Resource):

src/promptflow/promptflow/_sdk/_service/utils/utils.py

@@ -8,7 +8,7 @@
 from functools import wraps
 
 import psutil
-from flask import abort, request
+from flask import abort, make_response, request
 
 from promptflow._sdk._constants import DEFAULT_ENCODING, HOME_PROMPT_FLOW_DIR, PF_SERVICE_PORT_FILE
 from promptflow._sdk._errors import ConnectionNotFoundError, RunNotFoundError
@@ -85,6 +85,10 @@ def get_started_service_info(port):
     return service_info
 
 
+def make_response_no_content():
+    return make_response("", 204)
+
+
 @dataclass
 class ErrorInfo:
     exception: InitVar[Exception]

src/promptflow/tests/sdk_pfs_test/e2etests/test_connection_apis.py

@@ -45,6 +45,19 @@ def test_get_connection(self, pf_client: PFClient, pfs_op: PFSOperations) -> Non
             conn_from_pfs = pfs_op.get_connection_with_secret(name=name, status_code=200).json
         assert not conn_from_pfs["secrets"]["api_key"].startswith("*")
 
+    def test_delete_connection(self, pf_client: PFClient, pfs_op: PFSOperations) -> None:
+        len_connections = len(pfs_op.list_connections().json)
+
+        name = create_custom_connection(pf_client)
+        with check_activity_end_telemetry(
+            expected_activities=[
+                {"activity_name": "pf.connections.delete", "first_call": True},
+            ]
+        ):
+            pfs_op.delete_connection(name=name, status_code=204)
+        len_connections_after = len(pfs_op.list_connections().json)
+        assert len_connections_after == len_connections
+
     def test_list_connection_with_invalid_user(self, pfs_op: PFSOperations) -> None:
         # TODO: should we record telemetry for this case?
         with check_activity_end_telemetry(expected_activities=[]):

src/promptflow/tests/sdk_pfs_test/e2etests/test_run_apis.py

@@ -11,6 +11,7 @@
 
 from promptflow import PFClient
 from promptflow._sdk.entities import Run
+from promptflow._sdk.operations._local_storage_operations import LocalStorageOperations
 from promptflow.contracts._run_management import RunMetadata
 
 from ..utils import PFSOperations, check_activity_end_telemetry
@@ -81,6 +82,22 @@ def test_archive_restore_run(self, pf_client: PFClient, pfs_op: PFSOperations) -
         runs = pfs_op.list_runs().json
         assert any([item["name"] == run.name for item in runs])
 
+    def test_delete_run(self, pf_client: PFClient, pfs_op: PFSOperations) -> None:
+        run = create_run_against_multi_line_data(pf_client)
+        local_storage = LocalStorageOperations(run)
+        path = local_storage.path
+        assert path.exists()
+        with check_activity_end_telemetry(
+            expected_activities=[
+                {"activity_name": "pf.runs.get", "first_call": False},
+                {"activity_name": "pf.runs.delete"},
+            ]
+        ):
+            pfs_op.delete_run(name=run.name, status_code=204)
+        runs = pfs_op.list_runs().json
+        assert not any([item["name"] == run.name for item in runs])
+        assert not path.exists()
+
     def test_visualize_run(self, pfs_op: PFSOperations) -> None:
         with check_activity_end_telemetry(
             expected_activities=[

src/promptflow/tests/sdk_pfs_test/utils.py

@@ -72,6 +72,12 @@ def list_connections(self, status_code=None):
             assert status_code == response.status_code, response.text
         return response
 
+    def delete_connection(self, name: str, status_code=None):
+        response = self._client.delete(f"{self.CONNECTION_URL_PREFIX}/{name}", headers=self.remote_user_header())
+        if status_code:
+            assert status_code == response.status_code, response.text
+        return response
+
     def list_connections_by_provider(self, working_dir, status_code=None):
         response = self._client.get(
             f"{self.CONNECTION_URL_PREFIX}/",
@@ -151,6 +157,12 @@ def restore_run(self, name: str, status_code=None):
             assert status_code == response.status_code, response.text
         return response
 
+    def delete_run(self, name: str, status_code=None):
+        response = self._client.delete(f"{self.RUN_URL_PREFIX}/{name}")
+        if status_code:
+            assert status_code == response.status_code, response.text
+        return response
+
     def get_run_visualize(self, name: str, status_code=None):
         response = self._client.get(f"{self.RUN_URL_PREFIX}/{name}/visualize")
         if status_code: