fix: replace residual usage of click with typer (#1242)

deploy/sdk/src/dynamo/sdk/cli/serve_dynamo.py

@@ -25,7 +25,7 @@ import os
 import typing as t
 from typing import Any
 
-import click
+import typer
 import uvicorn
 import uvloop
 from fastapi.responses import StreamingResponse
@@ -87,46 +87,31 @@ def add_fastapi_routes(app, service, class_instance):
     return added_routes
 
 
-@click.command()
-@click.argument("bento_identifier", type=click.STRING, required=False, default=".")
-@click.option("--service-name", type=click.STRING, required=False, default="")
-@click.option(
-    "--runner-map",
-    type=click.STRING,
-    envvar="BENTOML_RUNNER_MAP",
-    help="JSON string of runners map, default sets to envars `BENTOML_RUNNER_MAP`",
-)
-@click.option(
-    "--worker-env", type=click.STRING, default=None, help="Environment variables"
-)
-@click.option(
-    "--worker-id",
-    required=False,
-    type=click.INT,
-    default=None,
-    help="If set, start the server as a bare worker with the given worker ID. Otherwise start a standalone server with a supervisor process.",
-)
-@click.option(
-    "--custom-component-name",
-    required=False,
-    type=click.STRING,
-    default=None,
-    help="If set, use this custom component name instead of the default service name",
-)
-@click.option(
-    "--target",
-    type=click.STRING,
-    default="dynamo",
-    help="Specify the target: 'dynamo' or 'bento'.",
-)
+app = typer.Typer()
+
+
+@app.command()
 def main(
-    bento_identifier: str,
-    service_name: str,
-    runner_map: str | None,
-    worker_env: str | None,
-    worker_id: int | None,
-    custom_component_name: str | None,
-    target: str,
+    bento_identifier: str = typer.Argument(".", help="The bento identifier"),
+    service_name: str = typer.Option("", help="Service name"),
+    runner_map: str = typer.Option(
+        None,
+        envvar="BENTOML_RUNNER_MAP",
+        help="JSON string of runners map, default sets to envars `BENTOML_RUNNER_MAP`",
+    ),
+    worker_env: str = typer.Option(None, help="Environment variables"),
+    worker_id: int = typer.Option(
+        None,
+        help="If set, start the server as a bare worker with the given worker ID. Otherwise start a standalone server with a supervisor process.",
+    ),
+    custom_component_name: str = typer.Option(
+        None,
+        help="If set, use this custom component name instead of the default service name",
+    ),
+    target: str = typer.Option(
+        "dynamo",
+        help="Specify the target: 'dynamo' or 'bento'.",
+    ),
 ) -> None:
     # hack to avoid bentoml from respawning the workers after their leases are revoked
     os.environ["BENTOML_CONTAINERIZED"] = "true"
@@ -367,4 +352,4 @@ def main(
 
 
 if __name__ == "__main__":
-    main()
+    app()

deploy/sdk/src/dynamo/sdk/cli/utils.py

@@ -27,10 +27,8 @@ import random
 import socket
 from typing import Any, DefaultDict, Dict, Iterator, Optional, Protocol, TextIO, Union
 
-import click
 import typer
 import yaml
-from click import Command, Context
 from rich.console import Console
 
 from dynamo.planner.defaults import PlannerDefaults  # type: ignore[attr-defined]
@@ -61,62 +59,6 @@ class ServiceProtocol(Protocol):
         ...
 
 
-class DynamoCommandGroup(click.Group):
-    """Simplified version of BentoMLCommandGroup for Dynamo CLI"""
-
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
-        self.aliases = kwargs.pop("aliases", [])
-        super().__init__(*args, **kwargs)
-        self._commands: dict[str, list[str]] = {}
-        self._aliases: dict[str, str] = {}
-
-    def add_command(self, cmd: Command, name: str | None = None) -> None:
-        assert cmd.callback is not None
-        callback = cmd.callback
-        cmd.callback = callback
-        cmd.context_settings["max_content_width"] = 120
-        aliases = getattr(cmd, "aliases", None)
-        if aliases:
-            assert cmd.name
-            self._commands[cmd.name] = aliases
-            self._aliases.update({alias: cmd.name for alias in aliases})
-        return super().add_command(cmd, name)
-
-    def add_subcommands(self, group: click.Group) -> None:
-        if not isinstance(group, click.MultiCommand):
-            raise TypeError(
-                "DynamoCommandGroup.add_subcommands only accepts click.MultiCommand"
-            )
-        if isinstance(group, DynamoCommandGroup):
-            # Common wrappers are already applied, call the super() method
-            for name, cmd in group.commands.items():
-                super().add_command(cmd, name)
-            self._commands.update(group._commands)
-            self._aliases.update(group._aliases)
-        else:
-            for name, cmd in group.commands.items():
-                self.add_command(cmd, name)
-
-    def resolve_alias(self, cmd_name: str):
-        return self._aliases[cmd_name] if cmd_name in self._aliases else cmd_name
-
-    def get_command(self, ctx: Context, cmd_name: str) -> Command | None:
-        cmd_name = self.resolve_alias(cmd_name)
-        return super().get_command(ctx, cmd_name)
-
-    def add_single_command(self, group: click.Group, command_name: str) -> None:
-        """Add a single command from a group by name."""
-        if not isinstance(group, click.MultiCommand):
-            raise TypeError("Only accepts click.MultiCommand")
-
-        ctx = click.Context(group)
-        cmd = group.get_command(ctx, command_name)
-        if cmd is None:
-            raise ValueError(f"Command '{command_name}' not found in group")
-
-        self.add_command(cmd, command_name)
-
-
 @contextlib.contextmanager
 def reserve_free_port(
     host: str = "localhost",