[SDK/CLI] Add pfazure flow update (#2082)

# Description

[SDK/CLI] Support `pfazure flow update` to update flow's metadata like
`display_name`, `description` or `tags`.

# All Promptflow Contribution checklist:
- [x] **The pull request does not introduce [breaking changes].**
- [x] **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.

docs/reference/pfazure-command-reference.md

@@ -19,6 +19,7 @@ Manage flows.
 | Command | Description |
 | --- | --- |
 | [pfazure flow create](#pfazure-flow-create) | Create a flow. |
+| [pfazure flow update](#pfazure-flow-update) | Update a flow. |
 | [pfazure flow list](#pfazure-flow-list) | List flows in a workspace. |
 
 
@@ -43,8 +44,50 @@ Local path to the flow directory.
 `--set`
 
 Update an object by specifying a property path and value to set.
-- `display_name`: Flow display name that will be created in remote. Default to be flow folder name + timestamp if not specified.
-- `type`: Flow type. Default to be "standard" if not specified. Available types are: "standard", "evaluation", "chat".
+- `display_name`: Flow display name that will be created in remote. Default to be flow folder name + timestamp if not specified. e.g. "--set display_name=\<display_name\>".
+- `type`: Flow type. Default to be "standard" if not specified. Available types are: "standard", "evaluation", "chat". e.g. "--set type=\<type\>".
+- `description`: Flow description. e.g. "--set description=\<description\>."
+- `tags`: Flow tags. e.g. "--set tags.key1=value1 tags.key2=value2."
+
+`--subscription`
+
+Subscription id, required when there is no default value from `az configure`.
+
+`--resource-group -g`
+
+Resource group name, required when there is no default value from `az configure`.
+
+`--workspace-name -w`
+
+Workspace name, required when there is no default value from `az configure`.
+
+
+
+### pfazure flow update
+
+Update a flow's metadata, such as `display name`, `description` and `tags`.
+
+```bash
+pfazure flow update --flow
+                    [--set]
+                    [--subscription]
+                    [--resource-group]
+                    [--workspace-name]
+```
+
+#### Parameters
+
+`--flow`
+
+The flow name on azure. It's a guid that can be found from 2 ways:
+- After creating a flow to azure, it can be found in the printed message in "name" attribute.
+- Open a flow in azure portal, the guid is in the url. e.g. https://ml.azure.com/prompts/flow/{workspace-id}/{flow-name}/xxx
+
+
+`--set`
+
+Update an object by specifying a property path and value to set.
+- `display_name`: Flow display name. e.g. "--set display_name=\<display_name\>".
 - `description`: Flow description. e.g. "--set description=\<description\>."
 - `tags`: Flow tags. e.g. "--set tags.key1=value1 tags.key2=value2."
 

src/promptflow/CHANGELOG.md

@@ -9,6 +9,7 @@
   PF_USE_AZURE_CLI_CREDENTIAL=true
   ```
 - [SDK/CLI] Support setting timeout for `pfazure run stream`.
+- [SDK/CLI] Support `pfazure flow update` to update flow's metadata like `display_name`, `description` or `tags`.
 
 ### Bugs Fixed
 

src/promptflow/promptflow/_cli/_pf_azure/_flow.py

@@ -22,7 +22,8 @@
     _set_workspace_argument_for_subparsers,
     activate_action,
 )
-from promptflow._sdk._constants import get_list_view_type
+from promptflow._sdk._constants import AzureFlowSource, get_list_view_type
+from promptflow.azure._entities._flow import Flow
 
 
 def add_parser_flow(subparsers):
@@ -34,6 +35,7 @@ def add_parser_flow(subparsers):
     )
     flow_subparsers = flow_parser.add_subparsers()
     add_parser_flow_create(flow_subparsers)
+    add_parser_flow_update(flow_subparsers)
     add_parser_flow_show(flow_subparsers)
     add_parser_flow_list(flow_subparsers)
     flow_parser.set_defaults(action="flow")
@@ -63,9 +65,9 @@ def add_parser_flow_create(subparsers):
         "--flow", type=str, help="Source folder of the flow."
     )
     add_params = [
-        _set_workspace_argument_for_subparsers,
         add_param_source,
         add_param_set,
+        _set_workspace_argument_for_subparsers,
     ] + base_params
 
     activate_action(
@@ -79,6 +81,43 @@ def add_parser_flow_create(subparsers):
     )
 
 
+def add_parser_flow_update(subparsers):
+    """Add flow update parser to the pf flow subparsers."""
+    epilog = """
+Use "--set" to set flow properties that you want to update. Supported properties are: [display_name, description, tags].
+
+Note:
+    1. In "--set" parameter, if the key name consists of multiple words, use snake-case instead of kebab-case. e.g. "--set display_name=<flow-display-name>"
+    2. Parameter flow is required to update a flow. It's a guid that can be found from 2 ways:
+        a. After creating a flow to azure, it can be found in the printed message in "name" attribute.
+        b. Open a flow in azure portal, the guid is in the url. e.g. https://ml.azure.com/prompts/flow/<workspace-id>/<flow-name>/xxx
+
+Examples:
+
+# Update a flow display name
+pfazure flow update --flow <flow-name> --set display_name=<flow-display-name>
+"""  # noqa: E501
+
+    add_param_source = lambda parser: parser.add_argument(  # noqa: E731
+        "--flow", type=str, help="Flow name to be updated which is a guid."
+    )
+    add_params = [
+        add_param_source,
+        add_param_set,
+        _set_workspace_argument_for_subparsers,
+    ] + base_params
+
+    activate_action(
+        name="update",
+        description="A CLI tool to update a flow's metadata on Azure.",
+        epilog=epilog,
+        add_params=add_params,
+        subparsers=subparsers,
+        help_message="Update a flow's metadata on azure.",
+        action_param_name="sub_action",
+    )
+
+
 def add_parser_flow_list(subparsers):
     """Add flow list parser to the pf flow subparsers."""
     epilog = """
@@ -173,6 +212,8 @@ def dispatch_flow_commands(args: argparse.Namespace):
         show_flow(args)
     elif args.sub_action == "list":
         list_flows(args)
+    elif args.sub_action == "update":
+        update_flow(args)
 
 
 def _get_flow_operation(subscription_id, resource_group, workspace_name):
@@ -187,6 +228,14 @@ def create_flow(args: argparse.Namespace):
     pf.flows.create_or_update(flow=args.flow, **params)
 
 
+def update_flow(args: argparse.Namespace):
+    """Update a flow for promptflow."""
+    pf = _get_azure_pf_client(args.subscription, args.resource_group, args.workspace_name, debug=args.debug)
+    params = _parse_flow_metadata_args(args.params_override)
+    flow_object = Flow(name=args.flow, flow_source=AzureFlowSource.PF_SERVICE)
+    pf.flows.create_or_update(flow=flow_object, **params)
+
+
 def show_flow(args: argparse.Namespace):
     """Get a flow for promptflow."""
     pf = _get_azure_pf_client(args.subscription, args.resource_group, args.workspace_name, debug=args.debug)

src/promptflow/promptflow/azure/_entities/_flow.py

@@ -30,7 +30,7 @@ class Flow(AdditionalIncludesMixin):
 
     def __init__(
         self,
-        path: Union[str, PathLike],
+        path: Optional[Union[str, PathLike]] = None,
         name: Optional[str] = None,
         type: Optional[str] = None,
         description: Optional[str] = None,

src/promptflow/promptflow/azure/_restclient/flow_service_caller.py

@@ -170,6 +170,31 @@ def create_flow(
             **kwargs,
         )
 
+    @_request_wrapper()
+    def update_flow(
+        self,
+        subscription_id,  # type: str
+        resource_group_name,  # type: str
+        workspace_name,  # type: str
+        flow_id,  # type: str
+        experiment_id=None,  # type: str
+        body=None,  # type: Optional["_models.UpdateFlowRequest"]
+        **kwargs,  # type: Any
+    ):
+        headers = self._get_headers()
+        return self.caller.flows.update_flow(
+            subscription_id=subscription_id,
+            resource_group_name=resource_group_name,
+            workspace_name=workspace_name,
+            flow_id=flow_id,
+            # experiment id equals to the workspace id, this is a hard code logic whether to be done at sdk side
+            # or PFS side, and won't be changed in the foreseeable future. So we hard code it here.
+            experiment_id=self._workspace._workspace_id,
+            body=body,
+            headers=headers,
+            **kwargs,
+        )
+
     @_request_wrapper()
     def create_component_from_flow(
         self,

src/promptflow/promptflow/azure/operations/_flow_operations.py

@@ -90,14 +90,24 @@ def _index_service_endpoint_url(self):
         return endpoint + "index/v1.0" + self._service_caller._common_azure_url_pattern
 
     @monitor_operation(activity_name="pfazure.flows.create_or_update", activity_type=ActivityType.PUBLICAPI)
-    def create_or_update(self, flow: Union[str, Path], display_name=None, type=None, **kwargs) -> Flow:
+    def create_or_update(
+        self, flow: Union[str, Path, Flow], display_name: str = None, type: str = None, **kwargs
+    ) -> Flow:
         """Create a flow to remote from local source, or update the metadata of an existing flow.
 
-        .. note::
-            Functionality of updating flow metadata is yet to be supported.
+        .. admonition::  Update a flow
 
-        :param flow: The source of the flow to create.
-        :type flow: Union[str, Path]
+            To update an existing flow, you can only update the display name, description, and tags of the flow.
+            The flow name is a guid that can be found from 2 ways:
+
+            - After creating a flow to azure, it can be found in the printed message in "name" attribute.
+            - Open a flow in azure portal, the guid is in the url. e.g. ``https://ml.azure.com/prompts/flow/<workspace-id>/<flow-name>/xxx``
+
+
+        :param flow: The source of the flow to create or update. When creating a flow, fill with the local flow path.
+            When updating a flow, fill with the flow object with a valid flow name, see above docstring about how to
+            find a flow name on azure.
+        :type flow: Union[str, Path, ~promptflow.azure.entities.Flow]
         :param display_name: The display name of the flow to create. Default to be flow folder name + timestamp
             if not specified. e.g. "web-classification-10-27-2023-14-19-10"
         :type display_name: str
@@ -108,8 +118,21 @@ def create_or_update(self, flow: Union[str, Path], display_name=None, type=None,
         :type description: str
         :param tags: The tags of the flow to create. Default to be the tags in flow yaml file.
         :type tags: Dict[str, str]
-        """
+        """  # noqa: E501
+        # if the flow is a flow object with name specified, try to update the flow
+        if isinstance(flow, Flow):
+            if flow.name:
+                logger.info(f"Updating azure flow {flow.name!r}.")
+                return self._update_azure_flow(flow=flow, display_name=display_name, **kwargs)
+            else:
+                raise FlowOperationError(
+                    "Flow name is required to update a flow. please refer to 'pfazure flow update --help' "
+                    "to learn how to get the flow name on azure"
+                )
+
+        logger.info(f"Creating flow from local source {flow!r}.")
         # validate the parameters
+        flow = Path(flow).resolve()
         azure_flow, flow_display_name, flow_type, kwargs = FlowOperations._validate_flow_creation_parameters(
             flow, display_name, type, **kwargs
         )
@@ -132,6 +155,53 @@ def create_or_update(self, flow: Union[str, Path], display_name=None, type=None,
 
         return result_flow
 
+    def _update_azure_flow(self, flow: Flow, display_name, **kwargs):
+        """Update an existing flow in azure."""
+        logger.info("Validating flow update parameters.")
+
+        display_name = display_name or flow.display_name
+        if not isinstance(display_name, str) and display_name is not None:
+            logger.warn(f"Display name must be a string, got {display_name!r}.")
+            display_name = None
+
+        description = kwargs.get("description", flow.description)
+        if not isinstance(description, str) and description is not None:
+            logger.warn(f"Description must be a string, got {description!r}.")
+            description = None
+
+        tags = kwargs.get("tags", flow.tags)
+        if not isinstance(tags, dict) and tags is not None:
+            logger.warn(f"Tags must be a dictionary, got {tags!r}.")
+            tags = None
+
+        body = {
+            "flow_name": display_name,
+            "description": description,
+            "tags": tags,
+        }
+        body = {k: v for k, v in body.items() if v is not None}
+        logger.debug(f"Updating flow {flow.name!r} with data {body}.")
+
+        try:
+            self._service_caller.update_flow(
+                subscription_id=self._operation_scope.subscription_id,
+                resource_group_name=self._operation_scope.resource_group_name,
+                workspace_name=self._operation_scope.workspace_name,
+                flow_id=flow.name,
+                body=body,
+            )
+        except Exception as e:
+            raise FlowOperationError(
+                f"Failed to update azure flow {flow.name!r} due to: {str(e)}. If the flow is not found in azure, "
+                f"please make sure the flow name is correct."
+            ) from e
+
+        updated_flow = self.get(flow.name)
+        flow_dict = updated_flow._to_dict()
+        logger.info(f"Flow updated successfully:\n{json.dumps(flow_dict, indent=4)}")
+
+        return updated_flow
+
     @staticmethod
     def _validate_flow_creation_parameters(source, flow_display_name=None, flow_type=None, **kwargs):
         """Validate the parameters for flow creation operation."""

src/promptflow/tests/sdk_cli_azure_test/e2etests/test_flow_operations.py

@@ -7,6 +7,7 @@
 import pytest
 
 from promptflow.azure._entities._flow import Flow
+from promptflow.exceptions import UserErrorException
 
 from .._azure_utils import DEFAULT_TEST_TIMEOUT, PYTEST_TIMEOUT_METHOD
 from ..recording_utilities import is_live
@@ -23,6 +24,7 @@
     "single_worker_thread_pool",
     "vcr_recording",
 )
+@pytest.mark.xdist_group(name="pfazure_flow")
 class TestFlow:
     def test_create_flow(self, created_flow: Flow):
         # most of the assertions are in the fixture itself
@@ -36,6 +38,25 @@ def test_get_flow(self, pf, created_flow: Flow):
         for attr in attributes:
             assert getattr(result, attr) == getattr(created_flow, attr), f"Assertion failed for attribute: {attr!r}"
 
+    def test_update_flow(self, pf, created_flow: Flow):
+
+        test_meta = {
+            "display_name": "SDK test flow",
+            "description": "SDK test flow description",
+            "tags": {"owner": "sdk-test", "key1": "value1"},
+        }
+        # update flow
+        updated_flow = pf.flows.create_or_update(flow=created_flow, **test_meta)
+
+        assert updated_flow.display_name == test_meta["display_name"]
+        assert updated_flow.description == test_meta["description"]
+        assert updated_flow.tags == test_meta["tags"]
+
+        # test update with wrong flow id
+        with pytest.raises(UserErrorException, match=r"Flow with id fake_flow_name not found"):
+            updated_flow.name = "fake_flow_name"
+            pf.flows.create_or_update(updated_flow, display_name="A new test flow")
+
     @pytest.mark.skipif(
         condition=not is_live(),
         reason="Complicated test combining `pf flow test` and global config",

src/promptflow/tests/sdk_cli_azure_test/unittests/test_cli.py

@@ -3,7 +3,7 @@
 import sys
 from pathlib import Path
 from typing import List
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, patch
 
 import pandas as pd
 import pytest
@@ -281,6 +281,29 @@ def test_flow_create_with_unknown_field(self, mocker: MockFixture, operation_sco
         )
         mocked.assert_called_with(flow=flow_dir, random_key="random_value")
 
+    def test_flow_update(self, mocker: MockFixture, operation_scope_args):
+        from promptflow.azure.operations._flow_operations import FlowOperations
+
+        mocked = mocker.patch.object(FlowOperations, "_update_azure_flow")
+        mocked.return_value._to_dict.return_value = {"name": "test_run"}
+        run_pf_command(
+            "flow",
+            "update",
+            "--flow",
+            "test_flow",
+            "--set",
+            "display_name=test_flow_display_name",
+            "description='test_description'",
+            "tags.key1=value1",
+            *operation_scope_args,
+        )
+        mocked.assert_called_with(
+            flow=ANY,
+            display_name="test_flow_display_name",
+            description="test_description",
+            tags={"key1": "value1"},
+        )
+
     def test_flow_list(
         self,
         mocker: MockFixture,