feat(framework): Update control servicer for flwr run with app ID (#6085)

Co-authored-by: Heng Pan <pan@flower.ai>

Author: yan-gao-GY

framework/py/flwr/cli/app_cmd/review.py

@@ -36,6 +36,7 @@
     load_private_key,
     sign_message,
 )
+from flwr.supercore.utils import parse_app_spec, request_download_link
 
 from ..auth_plugin.oidc_cli_plugin import OidcCliPlugin
 from ..config_utils import (
@@ -45,7 +46,7 @@
 )
 from ..constant import FEDERATION_CONFIG_HELP_MESSAGE
 from ..install import install_from_fab
-from ..utils import load_cli_auth_plugin, parse_app_spec, request_download_link
+from ..utils import load_cli_auth_plugin
 
 TRY_AGAIN_MESSAGE = "Please try again or press CTRL+C to abort.\n"
 
@@ -104,12 +105,21 @@ def review(
     token = auth_plugin.access_token
 
     # Validate app version and ID format
-    app_id, app_version = parse_app_spec(app_spec)
+    try:
+        app_id, app_version = parse_app_spec(app_spec)
+    except ValueError as e:
+        typer.secho(f"❌ {e}", fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=1) from e
 
     # Download FAB
     typer.secho("Downloading FAB... ", fg=typer.colors.BLUE)
     url = f"{PLATFORM_API_URL}/hub/fetch-fab"
-    presigned_url = request_download_link(app_id, app_version, url, "fab_url")
+    try:
+        presigned_url, _ = request_download_link(app_id, app_version, url, "fab_url")
+    except ValueError as e:
+        typer.secho(f"❌ {e}", fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=1) from e
+
     fab_bytes = _download_fab(presigned_url)
 
     # Unpack FAB

framework/py/flwr/cli/new/new.py

@@ -27,13 +27,12 @@
 import typer
 
 from flwr.supercore.constant import PLATFORM_API_URL
+from flwr.supercore.utils import parse_app_spec, request_download_link
 
 from ..utils import (
     is_valid_project_name,
-    parse_app_spec,
     prompt_options,
     prompt_text,
-    request_download_link,
     sanitize_project_name,
 )
 
@@ -205,7 +204,12 @@ def _download_zip_to_memory(presigned_url: str) -> io.BytesIO:
 def download_remote_app_via_api(app_spec: str) -> None:
     """Download App from Platform API."""
     # Validate app version and ID format
-    app_id, app_version = parse_app_spec(app_spec)
+    try:
+        app_id, app_version = parse_app_spec(app_spec)
+    except ValueError as e:
+        typer.secho(f"❌ {e}", fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=1) from e
+
     app_name = app_id.split("/")[1]
 
     project_dir = Path.cwd() / app_name
@@ -228,7 +232,11 @@ def download_remote_app_via_api(app_spec: str) -> None:
     )
     # Fetch ZIP downloading URL
     url = f"{PLATFORM_API_URL}/hub/fetch-zip"
-    presigned_url = request_download_link(app_id, app_version, url, "zip_url")
+    try:
+        presigned_url, _ = request_download_link(app_id, app_version, url, "zip_url")
+    except ValueError as e:
+        typer.secho(f"❌ {e}", fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=1) from e
 
     print(
         typer.style(

framework/py/flwr/cli/run/run.py

@@ -46,14 +46,10 @@
 from flwr.proto.control_pb2 import StartRunRequest  # pylint: disable=E0611
 from flwr.proto.control_pb2_grpc import ControlStub
 from flwr.supercore.constant import NOOP_FEDERATION
+from flwr.supercore.utils import parse_app_spec
 
 from ..log import start_stream
-from ..utils import (
-    flwr_cli_grpc_exc_handler,
-    init_channel,
-    load_cli_auth_plugin,
-    parse_app_spec,
-)
+from ..utils import flwr_cli_grpc_exc_handler, init_channel, load_cli_auth_plugin
 
 CONN_REFRESH_PERIOD = 60  # Connection refresh period for log streaming (seconds)
 
@@ -111,8 +107,15 @@ def run(
         app_spec = None
         if (app_str := str(app)).startswith("@"):
             # Validate app version and ID format
-            _ = parse_app_spec(app_str)
+            try:
+                _ = parse_app_spec(app_str)
+            except ValueError as e:
+                typer.secho(f"❌ {e}", fg=typer.colors.RED, err=True)
+                raise typer.Exit(code=1) from e
+
             app_spec = app_str
+            # Set `app` to current directory for credential storage
+            app = Path(".")
         is_remote_app = app_spec is not None
 
         typer.secho("Loading project configuration... ", fg=typer.colors.BLUE)
@@ -212,15 +215,7 @@ def _run_with_control_api(
                 f"🎊 Successfully started run {res.run_id}", fg=typer.colors.GREEN
             )
         else:
-            if is_remote_app:
-                typer.secho(
-                    "❌ Failed to start run. Please check that the provided "
-                    "app identifier (@account_name/app_name) is correct.",
-                    fg=typer.colors.RED,
-                    err=True,
-                )
-            else:
-                typer.secho("❌ Failed to start run", fg=typer.colors.RED, err=True)
+            typer.secho("❌ Failed to start run", fg=typer.colors.RED, err=True)
             raise typer.Exit(code=1)
 
         if output_format == CliOutputFormat.JSON:

framework/py/flwr/cli/utils.py

@@ -25,7 +25,6 @@
 
 import grpc
 import pathspec
-import requests
 import typer
 
 from flwr.common.constant import (
@@ -48,8 +47,6 @@
     create_channel,
     on_channel_state_change,
 )
-from flwr.common.version import package_version as flwr_version
-from flwr.supercore.constant import APP_ID_PATTERN, APP_VERSION_PATTERN
 
 from .auth_plugin import CliAuthPlugin, get_cli_plugin_class
 from .cli_account_auth_interceptor import CliAccountAuthInterceptor
@@ -547,94 +544,6 @@ def flwr_cli_grpc_exc_handler() -> Iterator[None]:  # pylint: disable=too-many-b
         raise
 
 
-def request_download_link(
-    app_id: str, app_version: str | None, in_url: str, out_url: str
-) -> str:
-    """Request a download link for the given app from the Flower platform API.
-
-    Parameters
-    ----------
-    app_id : str
-        The application identifier.
-    app_version : str | None
-        The application version, or None for latest.
-    in_url : str
-        The API endpoint URL.
-    out_url : str
-        The key name for the download URL in the response.
-
-    Returns
-    -------
-    str
-        The download URL for the application.
-
-    Raises
-    ------
-    typer.Exit
-        If connection fails, app not found, or API request fails.
-    """
-    headers = {
-        "Content-Type": "application/json",
-        "Accept": "application/json",
-    }
-    body = {
-        "app_id": app_id,  # send raw string of app_id
-        "app_version": app_version,
-        "flwr_version": flwr_version,
-    }
-    try:
-        resp = requests.post(in_url, headers=headers, data=json.dumps(body), timeout=20)
-    except requests.RequestException as e:
-        typer.secho(
-            f"Unable to connect to Platform API: {e}",
-            fg=typer.colors.RED,
-            err=True,
-        )
-        raise typer.Exit(code=1) from e
-
-    if resp.status_code == 404:
-        error_message = resp.json()["detail"]
-        if isinstance(error_message, dict):
-            available_app_versions = error_message["available_app_versions"]
-            available_versions_str = (
-                ", ".join(map(str, available_app_versions))
-                if available_app_versions
-                else "None"
-            )
-            typer.secho(
-                f"{app_id}=={app_version} not found in Platform API. "
-                f"Available app versions for {app_id}: {available_versions_str}",
-                fg=typer.colors.RED,
-                err=True,
-            )
-        else:
-            typer.secho(
-                f"{app_id} not found in Platform API.",
-                fg=typer.colors.RED,
-                err=True,
-            )
-        raise typer.Exit(code=1)
-
-    if not resp.ok:
-        typer.secho(
-            f"Platform API request failed with "
-            f"status {resp.status_code}. Details: {resp.text}",
-            fg=typer.colors.RED,
-            err=True,
-        )
-        raise typer.Exit(code=1)
-
-    data = resp.json()
-    if out_url not in data:
-        typer.secho(
-            "Invalid response from Platform API",
-            fg=typer.colors.RED,
-            err=True,
-        )
-        raise typer.Exit(code=1)
-    return str(data[out_url])
-
-
 def build_pathspec(patterns: Iterable[str]) -> pathspec.PathSpec:
     """Build a PathSpec from a list of GitIgnore-style patterns.
 
@@ -711,49 +620,3 @@ def validate_credentials_content(creds_path: Path) -> str:
         raise typer.Exit(code=1)
 
     return creds[ACCESS_TOKEN_KEY]
-
-
-def parse_app_spec(app_spec: str) -> tuple[str, str | None]:
-    """Parse app specification string into app ID and version.
-
-    Parameters
-    ----------
-    app_spec : str
-        The app specification string in the format '@account/app' or
-        '@account/app==x.y.z' (digits only).
-
-    Returns
-    -------
-    tuple[str, str | None]
-        A tuple containing the app ID and optional version.
-
-    Raises
-    ------
-    typer.Exit
-        If the app specification format is invalid.
-    """
-    if "==" in app_spec:
-        app_id, app_version = app_spec.split("==")
-
-        # Validate app version format
-        if not re.match(APP_VERSION_PATTERN, app_version):
-            typer.secho(
-                "❌ Invalid app version. Expected format: x.y.z (digits only).",
-                fg=typer.colors.RED,
-                err=True,
-            )
-            raise typer.Exit(code=1)
-    else:
-        app_id = app_spec
-        app_version = None
-
-    # Validate app_id format
-    if not re.match(APP_ID_PATTERN, app_id):
-        typer.secho(
-            "❌ Invalid remote app ID. Expected format: '@account/app'.",
-            fg=typer.colors.RED,
-            err=True,
-        )
-        raise typer.Exit(code=1)
-
-    return app_id, app_version

framework/py/flwr/cli/utils_test.py

@@ -22,14 +22,10 @@
 import unittest
 from pathlib import Path
 
-import pytest
-import typer
-
 from flwr.cli.utils import (
     build_pathspec,
     get_sha256_hash,
     load_gitignore_patterns,
-    parse_app_spec,
     validate_credentials_content,
 )
 from flwr.common.constant import (
@@ -162,23 +158,3 @@ def test_load_gitignore_patterns_with_pathspec() -> None:
 
     # Should not match normal files
     assert spec.match_file("good.py") is False
-
-
-@pytest.mark.parametrize(
-    "value",
-    [
-        "user/app==1.2.3",  # missing '@'
-        "@accountapp==1.2.3",  # missing slash
-        "@account/app==1.2",  # bad version
-        "@account/app==1.2.3.4",  # bad version
-        "@account*/app==1.2.3",  # bad user id chars
-        "@account/app*==1.2.3",  # bad app id chars
-    ],
-)
-def test_parse_app_spec_rejects_invalid_formats(value: str) -> None:
-    """For an invalid string, the function should fail fast with typer.Exit(code=1)."""
-    with pytest.raises(typer.Exit) as exc:
-        parse_app_spec(value)
-
-    # Ensure we specifically exited with code 1
-    assert exc.value.exit_code == 1

framework/py/flwr/server/app.py

@@ -324,6 +324,7 @@ def run_superlink() -> None:
         authz_plugin=authz_plugin,
         event_log_plugin=event_log_plugin,
         artifact_provider=artifact_provider,
+        fleet_api_type=args.fleet_api_type,
     )
     grpc_servers = [control_server]
     bckg_threads: list[threading.Thread] = []