Refactor CLI using dtyper (#306)

docs/docs/user/references/cli-commands.md

@@ -44,7 +44,7 @@ $ kpops clean [OPTIONS] PIPELINE_PATH [COMPONENTS_MODULE]
 * `--config FILE`: Path to the config.yaml file  [env var: KPOPS_CONFIG_PATH; default: config.yaml]
 * `--steps TEXT`: Comma separated list of steps to apply the command on  [env var: KPOPS_PIPELINE_STEPS]
 * `--dry-run / --execute`: Whether to dry run the command or execute it  [default: dry-run]
-* `--verbose / --no-verbose`: [default: no-verbose]
+* `--verbose / --no-verbose`: Enable verbose printing  [default: no-verbose]
 * `--help`: Show this message and exit.
 
 ## `kpops deploy`
@@ -67,8 +67,8 @@ $ kpops deploy [OPTIONS] PIPELINE_PATH [COMPONENTS_MODULE]
 * `--pipeline-base-dir DIRECTORY`: Base directory to the pipelines (default is current working directory)  [env var: KPOPS_PIPELINE_BASE_DIR; default: .]
 * `--defaults DIRECTORY`: Path to defaults folder  [env var: KPOPS_DEFAULT_PATH]
 * `--config FILE`: Path to the config.yaml file  [env var: KPOPS_CONFIG_PATH; default: config.yaml]
-* `--verbose / --no-verbose`: [default: no-verbose]
 * `--dry-run / --execute`: Whether to dry run the command or execute it  [default: dry-run]
+* `--verbose / --no-verbose`: Enable verbose printing  [default: no-verbose]
 * `--steps TEXT`: Comma separated list of steps to apply the command on  [env var: KPOPS_PIPELINE_STEPS]
 * `--help`: Show this message and exit.
 
@@ -93,8 +93,8 @@ $ kpops destroy [OPTIONS] PIPELINE_PATH [COMPONENTS_MODULE]
 * `--defaults DIRECTORY`: Path to defaults folder  [env var: KPOPS_DEFAULT_PATH]
 * `--config FILE`: Path to the config.yaml file  [env var: KPOPS_CONFIG_PATH; default: config.yaml]
 * `--steps TEXT`: Comma separated list of steps to apply the command on  [env var: KPOPS_PIPELINE_STEPS]
+* `--verbose / --no-verbose`: Enable verbose printing  [default: no-verbose]
 * `--dry-run / --execute`: Whether to dry run the command or execute it  [default: dry-run]
-* `--verbose / --no-verbose`: [default: no-verbose]
 * `--help`: Show this message and exit.
 
 ## `kpops generate`
@@ -147,7 +147,7 @@ $ kpops reset [OPTIONS] PIPELINE_PATH [COMPONENTS_MODULE]
 * `--config FILE`: Path to the config.yaml file  [env var: KPOPS_CONFIG_PATH; default: config.yaml]
 * `--steps TEXT`: Comma separated list of steps to apply the command on  [env var: KPOPS_PIPELINE_STEPS]
 * `--dry-run / --execute`: Whether to dry run the command or execute it  [default: dry-run]
-* `--verbose / --no-verbose`: [default: no-verbose]
+* `--verbose / --no-verbose`: Enable verbose printing  [default: no-verbose]
 * `--help`: Show this message and exit.
 
 ## `kpops schema`

kpops/__init__.py

@@ -1 +1,12 @@
 __version__ = "1.3.2"
+
+# export public API functions
+from kpops.cli.main import clean, deploy, destroy, generate, reset
+
+__all__ = (
+    "generate",
+    "deploy",
+    "destroy",
+    "reset",
+    "clean",
+)

kpops/cli/main.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Iterator, Optional
 
+import dtyper
 import typer
 
 from kpops import __version__
@@ -25,7 +26,7 @@
 
 LOG_DIVIDER = "#" * 100
 
-app = typer.Typer(pretty_exceptions_enable=False)
+app = dtyper.Typer(pretty_exceptions_enable=False)
 
 BASE_DIR_PATH_OPTION: Path = typer.Option(
     default=Path("."),
@@ -77,6 +78,8 @@
     help="Whether to dry run the command or execute it",
 )
 
+VERBOSE_OPTION = typer.Option(False, help="Enable verbose printing")
+
 COMPONENTS_MODULES: str | None = typer.Argument(
     default=None,
     help="Custom Python module containing your project-specific components",
@@ -160,7 +163,9 @@ def get_steps_to_apply(
     return list(pipeline)
 
 
-def reverse_pipeline_steps(pipeline, steps) -> Iterator[PipelineComponent]:
+def reverse_pipeline_steps(
+    pipeline: Pipeline, steps: str | None
+) -> Iterator[PipelineComponent]:
     return reversed(get_steps_to_apply(pipeline, steps))
 
 
@@ -185,7 +190,7 @@ def create_pipeline_config(
     return pipeline_config
 
 
-@app.command(
+@app.command(  # pyright: ignore[reportGeneralTypeIssues] https://github.com/rec/dtyper/issues/8
     help="""
     Generate json schema.
 
@@ -215,16 +220,16 @@ def schema(
             gen_config_schema()
 
 
-@app.command(
+@app.command(  # pyright: ignore[reportGeneralTypeIssues] https://github.com/rec/dtyper/issues/8
     help="Enriches pipelines steps with defaults. The output is used as input for the deploy/destroy/... commands."
 )
 def generate(
-    pipeline_base_dir: Path = BASE_DIR_PATH_OPTION,
     pipeline_path: Path = PIPELINE_PATH_ARG,
     components_module: Optional[str] = COMPONENTS_MODULES,
+    pipeline_base_dir: Path = BASE_DIR_PATH_OPTION,
     defaults: Optional[Path] = DEFAULT_PATH_OPTION,
     config: Path = CONFIG_PATH_OPTION,
-    verbose: bool = typer.Option(False, help="Enable verbose printing"),
+    verbose: bool = VERBOSE_OPTION,
     template: bool = typer.Option(False, help="Run Helm template"),
     steps: Optional[str] = PIPELINE_STEPS,
     api_version: Optional[str] = typer.Option(
@@ -259,15 +264,17 @@ def generate(
     return pipeline
 
 
-@app.command(help="Deploy pipeline steps")
+@app.command(
+    help="Deploy pipeline steps"
+)  # pyright: ignore[reportGeneralTypeIssues] https://github.com/rec/dtyper/issues/8
 def deploy(
-    pipeline_base_dir: Path = BASE_DIR_PATH_OPTION,
     pipeline_path: Path = PIPELINE_PATH_ARG,
     components_module: Optional[str] = COMPONENTS_MODULES,
+    pipeline_base_dir: Path = BASE_DIR_PATH_OPTION,
     defaults: Optional[Path] = DEFAULT_PATH_OPTION,
     config: Path = CONFIG_PATH_OPTION,
-    verbose: bool = False,
     dry_run: bool = DRY_RUN,
+    verbose: bool = VERBOSE_OPTION,
     steps: Optional[str] = PIPELINE_STEPS,
 ):
     pipeline_config = create_pipeline_config(config, defaults, verbose)
@@ -281,16 +288,18 @@ def deploy(
         component.deploy(dry_run)
 
 
-@app.command(help="Destroy pipeline steps")
+@app.command(
+    help="Destroy pipeline steps"
+)  # pyright: ignore[reportGeneralTypeIssues] https://github.com/rec/dtyper/issues/8
 def destroy(
-    pipeline_base_dir: Path = BASE_DIR_PATH_OPTION,
     pipeline_path: Path = PIPELINE_PATH_ARG,
     components_module: Optional[str] = COMPONENTS_MODULES,
+    pipeline_base_dir: Path = BASE_DIR_PATH_OPTION,
     defaults: Optional[Path] = DEFAULT_PATH_OPTION,
     config: Path = CONFIG_PATH_OPTION,
     steps: Optional[str] = PIPELINE_STEPS,
+    verbose: bool = VERBOSE_OPTION,
     dry_run: bool = DRY_RUN,
-    verbose: bool = False,
 ):
     pipeline_config = create_pipeline_config(config, defaults, verbose)
     pipeline = setup_pipeline(
@@ -302,16 +311,18 @@ def destroy(
         component.destroy(dry_run)
 
 
-@app.command(help="Reset pipeline steps")
+@app.command(
+    help="Reset pipeline steps"
+)  # pyright: ignore[reportGeneralTypeIssues] https://github.com/rec/dtyper/issues/8
 def reset(
-    pipeline_base_dir: Path = BASE_DIR_PATH_OPTION,
     pipeline_path: Path = PIPELINE_PATH_ARG,
     components_module: Optional[str] = COMPONENTS_MODULES,
+    pipeline_base_dir: Path = BASE_DIR_PATH_OPTION,
     defaults: Optional[Path] = DEFAULT_PATH_OPTION,
     config: Path = CONFIG_PATH_OPTION,
     steps: Optional[str] = PIPELINE_STEPS,
     dry_run: bool = DRY_RUN,
-    verbose: bool = False,
+    verbose: bool = VERBOSE_OPTION,
 ):
     pipeline_config = create_pipeline_config(config, defaults, verbose)
     pipeline = setup_pipeline(
@@ -324,16 +335,18 @@ def reset(
         component.reset(dry_run)
 
 
-@app.command(help="Clean pipeline steps")
+@app.command(
+    help="Clean pipeline steps"
+)  # pyright: ignore[reportGeneralTypeIssues] https://github.com/rec/dtyper/issues/8
 def clean(
-    pipeline_base_dir: Path = BASE_DIR_PATH_OPTION,
     pipeline_path: Path = PIPELINE_PATH_ARG,
     components_module: Optional[str] = COMPONENTS_MODULES,
+    pipeline_base_dir: Path = BASE_DIR_PATH_OPTION,
     defaults: Optional[Path] = DEFAULT_PATH_OPTION,
     config: Path = CONFIG_PATH_OPTION,
     steps: Optional[str] = PIPELINE_STEPS,
     dry_run: bool = DRY_RUN,
-    verbose: bool = False,
+    verbose: bool = VERBOSE_OPTION,
 ):
     pipeline_config = create_pipeline_config(config, defaults, verbose)
     pipeline = setup_pipeline(

kpops/pipeline_generator/pipeline.py

@@ -56,6 +56,9 @@ def __bool__(self) -> bool:
     def __iter__(self) -> Iterator[PipelineComponent]:
         return iter(self.components)
 
+    def __len__(self) -> int:
+        return len(self.components)
+
     def validate_unique_names(self) -> None:
         step_names = [component.name for component in self.components]
         duplicates = [name for name, count in Counter(step_names).items() if count > 1]
@@ -287,6 +290,9 @@ def __str__(self) -> str:
             )
         )
 
+    def __len__(self) -> int:
+        return len(self.components)
+
     def substitute_in_component(self, component_as_dict: dict) -> dict:
         """Substitute all $-placeholders in a component in dict representation
 

pyproject.toml

@@ -32,6 +32,7 @@ pydantic = { extras = ["dotenv"], version = "^1.10.8" }
 rich = "^12.4.4"
 PyYAML = "^6.0"
 typer = { extras = ["all"], version = "^0.6.1" }
+dtyper = "^2.1.0"
 pyhumps = "^3.7.3"
 cachetools = "^5.2.0"
 dictdiffer = "^0.9.0"