Feat: switch cli to typer (#8)

* feat: switch cli to typer

* fix: add options short names + use enum value

README.md

@@ -116,13 +116,14 @@ vertex-deployer --help
 ## Backlog
 1. Features
     1. handle multiple config files formats (toml, json, yaml)
-    2. allow for multiple config files as inputs
+    2. allow for multiple config files as inputs -> give a connector to generate json ?
     3. CLI to typer to have multiple commands (check, deploy, init, etc)
     4. Possibility to store env variables in a env class stored somewhere
     5. Dynamic config checks using pydantic
-    6. Scheduling with cloud function instead of cloud scheduler
+    6. Scheduling with cloud function instead of Vertex scheduler API
     7. versioning of config files on a gcs bucket
+    8. compile with parameters option
 2. Add more documentation
 3. Add review flow
 4. Add PR and Issue templates
-5. publish on Pypi
+5. publish on Pypi (temp fix: .tar.gz file on gcs?)

deployer/cli.py

@@ -1,122 +1,116 @@
-import argparse
+import logging
 import os
 from pathlib import Path
 
+import typer
 from dotenv import find_dotenv, load_dotenv
+from typing_extensions import Annotated
 
-from deployer.constants import DEFAULT_LOCAL_PACKAGE_PATH, PIPELINE_ROOT_PATH
+from deployer.constants import (
+    DEFAULT_LOCAL_PACKAGE_PATH,
+    DEFAULT_TAGS,
+    PIPELINE_ROOT_PATH,
+)
 from deployer.deployer import VertexPipelineDeployer
 from deployer.utils import (
     import_pipeline_from_dir,
     load_config,
     make_pipeline_names_enum_from_dir,
 )
 
-PipelineNames = make_pipeline_names_enum_from_dir(PIPELINE_ROOT_PATH)
-
-
-def get_args():  # noqa: D103
-    parser = argparse.ArgumentParser()
-    parser.add_argument(
-        "pipeline_name",
-        type=str,
-        choices=[e.value for e in PipelineNames],
-        help="The name of the pipeline to run.",
-    )
-    parser.add_argument(
-        "--env-file",
-        "-e",
-        type=str,
-        default=argparse.SUPPRESS,
-        help="The environment to run the pipeline in.",
-    )
-    parser.add_argument(
-        "--compile",
-        "-c",
-        action="store_true",
-        help="Whether to compile the pipeline.",
-    )
-    parser.add_argument(
-        "--upload",
-        "-u",
-        action="store_true",
-        help="Whether to upload the pipeline to the registry.",
-    )
-    parser.add_argument(
-        "--run",
-        "-r",
-        action="store_true",
-        help="Whether to run the pipeline.",
-    )
-    parser.add_argument(
-        "--schedule",
-        "-s",
-        action="store_true",
-        help="Whether to create a schedule for the pipeline.",
-    )
-    parser.add_argument(
-        "--cron",
-        type=str,
-        default=argparse.SUPPRESS,
-        help="Cron expression for scheduling the pipeline.",
-    )
-    parser.add_argument(
-        "--delete-last-schedule",
-        "-dls",
-        action="store_true",
-        help="Whether to delete the previous schedule before creating a new one.",
-    )
-    parser.add_argument(
-        "--tags",
-        type=str,
-        nargs="*",
-        help="The tags to use when uploading the pipeline.",
-    )
-    parser.add_argument(
-        "--config-name",
-        "-cn",
-        type=str,
-        default=argparse.SUPPRESS,
-        help="The name of the configuration file to use when running the pipeline.",
-    )
-    parser.add_argument(
-        "--enable-caching",
-        "-ec",
-        action="store_true",
-        help="Whether to enable caching when running the pipeline.",
-    )
-    parser.add_argument(
-        "--experiment-name",
-        "-exp",
-        type=str,
-        default=argparse.SUPPRESS,
-        help="The name of the experiment to run the pipeline in.",
-    )
-    parser.add_argument(
-        "--local-package-path",
-        "-lpp",
-        type=Path,
-        default=argparse.SUPPRESS,
-        help="The path to the local package to upload.",
-    )
-    return parser.parse_args()
-
-
-def main(  # noqa: D103
-    pipeline_name: PipelineNames,
-    env_file: str | None = None,
-    compile: bool = True,
-    upload: bool = False,
-    run: bool = True,
-    schedule: bool = False,
-    cron: str | None = None,
-    delete_last_schedule: bool = False,
-    tags: list = ["latest"],  # noqa: B006
-    config_name: str | None = None,
-    enable_caching: bool = False,
-    experiment_name: str | None = None,
-    local_package_path: Path = DEFAULT_LOCAL_PACKAGE_PATH,
-) -> None:
+logging.basicConfig(level=logging.INFO, handlers=[logging.StreamHandler()])
+
+app = typer.Typer(no_args_is_help=True, rich_help_panel="rich")
+
+PipelineName = make_pipeline_names_enum_from_dir(PIPELINE_ROOT_PATH)
+
+
+@app.command(no_args_is_help=True)
+def deploy(
+    pipeline_name: Annotated[
+        PipelineName, typer.Argument(..., help="The name of the pipeline to run.")
+    ],
+    env_file: Annotated[
+        Path,
+        typer.Option(
+            help="The environment file to use.",
+            exists=True,
+            dir_okay=False,
+            file_okay=True,
+            resolve_path=True,
+        ),
+    ] = None,
+    compile: Annotated[
+        bool,
+        typer.Option("--compile/--no-compile", "-c/-nc", help="Whether to compile the pipeline."),
+    ] = True,
+    upload: Annotated[
+        bool,
+        typer.Option(
+            "--upload/--no-upload",
+            "-u/-nu",
+            help="Whether to upload the pipeline to Google Artifact Registry.",
+        ),
+    ] = False,
+    run: Annotated[
+        bool, typer.Option("--run/--no-run", "-r/-nr", help="Whether to run the pipeline.")
+    ] = False,
+    schedule: Annotated[
+        bool,
+        typer.Option(
+            "--schedule/--no-schedule",
+            "-s/-ns",
+            help="Whether to create a schedule for the pipeline.",
+        ),
+    ] = False,
+    cron: Annotated[str, typer.Option(help="Cron expression for scheduling the pipeline.")] = None,
+    delete_last_schedule: Annotated[
+        bool,
+        typer.Option(
+            "--delete-last-schedule",
+            "-dls",
+            help="Whether to delete the previous schedule before creating a new one.",
+        ),
+    ] = False,
+    tags: Annotated[
+        list[str], typer.Option(help="The tags to use when uploading the pipeline.")
+    ] = DEFAULT_TAGS,
+    config_name: Annotated[
+        str,
+        typer.Option(
+            "--config-name",
+            "-cn",
+            help="The name of the configuration file to use when running the pipeline.",
+        ),
+    ] = None,
+    enable_caching: Annotated[
+        bool,
+        typer.Option(
+            "--enable-caching", "-ec", help="Whether to enable caching when running the pipeline."
+        ),
+    ] = False,
+    experiment_name: Annotated[
+        str,
+        typer.Option(
+            "--experiment-name",
+            "-en",
+            help="The name of the experiment to run the pipeline in."
+            "Defaults to '{pipeline_name}-experiment'.",
+        ),
+    ] = None,
+    local_package_path: Annotated[
+        Path,
+        typer.Option(
+            "--local-package-path",
+            "-lpp",
+            help="The path to the local package to upload.",
+            dir_okay=True,
+            file_okay=False,
+            resolve_path=True,
+        ),
+    ] = DEFAULT_LOCAL_PACKAGE_PATH,
+):
+    """Deploy and manage Vertex AI Pipelines."""
     if env_file is not None:
         find_dotenv(env_file, raise_error_if_not_found=True)
         load_dotenv(env_file)
@@ -125,7 +119,9 @@ def main(  # noqa: D103
     region = os.environ["GCP_REGION"]
     staging_bucket_name = os.environ["VERTEX_STAGING_BUCKET_NAME"]
     service_account = os.environ["VERTEX_SERVICE_ACCOUNT"]
-    pipeline_func = import_pipeline_from_dir(PIPELINE_ROOT_PATH, pipeline_name.replace("-", "_"))
+    pipeline_func = import_pipeline_from_dir(
+        PIPELINE_ROOT_PATH, pipeline_name.value.replace("-", "_")
+    )
     gar_location = os.environ["GAR_LOCATION"] if (upload or schedule) else None
     gar_repo_id = f"{os.environ['GAR_REPO_ID']}-kfp" if (upload or schedule) else None
 
@@ -134,7 +130,7 @@ def main(  # noqa: D103
         region=region,
         staging_bucket_name=staging_bucket_name,
         service_account=service_account,
-        pipeline_name=pipeline_name,
+        pipeline_name=pipeline_name.value,
         pipeline_func=pipeline_func,
         gar_location=gar_location,
         gar_repo_id=gar_repo_id,
@@ -172,14 +168,5 @@ def main(  # noqa: D103
             enable_caching=enable_caching,
             parameter_values=SELECTED_CONFIGURATION,
             tag=tags[0] if tags else None,
+            delete_last_schedule=delete_last_schedule,
         )
-
-
-def cli():  # noqa: D103
-    import logging
-
-    logging.basicConfig(level=logging.INFO, handlers=[logging.StreamHandler()])
-
-    args = get_args()
-
-    main(**vars(args))

deployer/constants.py

@@ -3,3 +3,4 @@
 
 DEFAULT_SCHEDULER_TIMEZONE = "Europe/Paris"
 DEFAULT_LOCAL_PACKAGE_PATH = "vertex/pipelines/compiled_pipelines"
+DEFAULT_TAGS = ["latest"]

pyproject.toml

@@ -7,14 +7,16 @@ readme = "README.md"
 packages = [{include = "deployer"}]
 
 [tool.poetry.scripts]
-vertex-deployer = "deployer.cli:cli"
+vertex-deployer = "deployer.cli:app"
 
 [tool.poetry.dependencies]
 python = ">=3.10,<3.11.0"
 python-dotenv = "^1.0.0"
 kfp = ">=2.0.1, <2.1.0"
 google-cloud-aiplatform = "^1.26.1"
 requests = "^2.31.0"
+typer = "^0.9.0"
+rich = {version = "^13.5.3", optional = true}
 
 [tool.poetry.group.dev.dependencies]
 black = "^23.7.0"
@@ -25,6 +27,9 @@ ipykernel = "6.9.1"
 nbstripout = "^0.6.1"
 ruff = "^0.0.289"
 
+[tool.poetry.extras]
+rich = ["rich"]
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"
@@ -50,7 +55,7 @@ select = ["B", "C", "D", "E", "F", "W"]
 convention = "google"
 
 [tool.ruff.per-file-ignores]
-"*run.py" = ["D"]
+"*cli.py" = ["D", "B008"]
 "*__init__.py" = [
     "F401",
     "D100",