Create init command

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

@@ -19,6 +19,7 @@ $ kpops [OPTIONS] COMMAND [ARGS]...
 * `deploy`: Deploy pipeline steps
 * `destroy`: Destroy pipeline steps
 * `generate`: Generate enriched pipeline representation
+* `init`: Initialize a new KPOps project.
 * `manifest`: Render final resource representation
 * `reset`: Reset pipeline steps
 * `schema`: Generate JSON schema.
@@ -126,6 +127,25 @@ $ kpops generate [OPTIONS] PIPELINE_PATH
 * `--verbose / --no-verbose`: Enable verbose printing  [default: no-verbose]
 * `--help`: Show this message and exit.
 
+## `kpops init`
+
+Initialize a new KPOps project.
+
+**Usage**:
+
+```console
+$ kpops init [OPTIONS] PATH
+```
+
+**Arguments**:
+
+* `PATH`: Path for a new KPOps project. It should lead to an empty (or non-existent) directory. The part of the path that doesn't exist will be created.  [required]
+
+**Options**:
+
+* `--config-include-opt / --no-config-include-opt`: Whether to include non-required settings in the generated 'config.yaml'  [default: no-config-include-opt]
+* `--help`: Show this message and exit.
+
 ## `kpops manifest`
 
 In addition to generate, render final resource representation for each pipeline step, e.g. Kubernetes manifests.

hooks/gen_docs/gen_docs_env_vars.py

@@ -228,7 +228,6 @@ def write_csv_to_md_file(
     :param source: path to csv file to read from
     :param target: path to md file to overwrite or create
     :param title: Title for the table, optional
-
     """
     if heading:
         heading += " "

kpops/__init__.py

@@ -1,7 +1,7 @@
 __version__ = "4.0.1"
 
 # export public API functions
-from kpops.cli.main import clean, deploy, destroy, generate, manifest, reset
+from kpops.cli.main import clean, deploy, destroy, generate, init, manifest, reset
 
 __all__ = (
     "generate",
@@ -10,4 +10,5 @@
     "destroy",
     "reset",
     "clean",
+    "init",
 )

kpops/cli/main.py

@@ -22,6 +22,7 @@
 from kpops.components.base_components.models.resource import Resource
 from kpops.config import ENV_PREFIX, KpopsConfig
 from kpops.pipeline import ComponentFilterPredicate, Pipeline, PipelineGenerator
+from kpops.utils.cli_commands import init_project
 from kpops.utils.gen_schema import (
     SchemaScope,
     gen_config_schema,
@@ -71,7 +72,22 @@
     help="Path to YAML with pipeline definition",
 )
 
-PIPELINE_STEPS: str | None = typer.Option(
+PROJECT_PATH: Path = typer.Argument(
+    default=...,
+    exists=False,
+    file_okay=False,
+    dir_okay=True,
+    readable=True,
+    resolve_path=True,
+    help="Path for a new KPOps project. It should lead to an empty (or non-existent) directory. The part of the path that doesn't exist will be created.",
+)
+
+CONFIG_INCLUDE_OPTIONAL: bool = typer.Option(
+    default=False,
+    help="Whether to include non-required settings in the generated 'config.yaml'",
+)
+
+PIPELINE_STEPS: Optional[str] = typer.Option(
     default=None,
     envvar=f"{ENV_PREFIX}PIPELINE_STEPS",
     help="Comma separated list of steps to apply the command on",
@@ -109,6 +125,7 @@
     ),
 )
 
+
 logger = logging.getLogger()
 logging.getLogger("httpx").setLevel(logging.WARNING)
 stream_handler = logging.StreamHandler()
@@ -189,6 +206,21 @@ def create_kpops_config(
     )
 
 
+@app.command(  # pyright: ignore[reportCallIssue] https://github.com/rec/dtyper/issues/8
+    help="Initialize a new KPOps project."
+)
+def init(
+    path: Path = PROJECT_PATH,
+    config_include_opt: bool = CONFIG_INCLUDE_OPTIONAL,
+):
+    if not path.exists():
+        path.mkdir(parents=False)
+    elif next(path.iterdir(), False):
+        log.warning("Please provide a path to an empty directory.")
+        return
+    init_project(path, config_include_opt)
+
+
 @app.command(  # pyright: ignore[reportCallIssue] https://github.com/rec/dtyper/issues/8
     help="""
     Generate JSON schema.

kpops/utils/cli_commands.py

@@ -0,0 +1,81 @@
+import logging
+from pathlib import Path
+from typing import Any
+
+import yaml
+from pydantic import BaseModel
+from pydantic.fields import FieldInfo
+from pydantic_core import PydanticUndefined
+
+from hooks.gen_docs.gen_docs_env_vars import collect_fields
+from kpops.config import KpopsConfig
+from kpops.utils.docstring import describe_object
+from kpops.utils.json import is_jsonable
+from kpops.utils.pydantic import issubclass_patched
+
+log = logging.getLogger("cli_commands_utils")
+
+
+def extract_config_fields_for_yaml(
+    fields: dict[str, Any], required: bool
+) -> dict[str, Any]:
+    """Return only (non-)required fields and their respective default values.
+
+    :param fields: Dict containing the fields to be categorized. The key of a
+        record is the name of the field, the value is the field's type.
+    :param required: Whether to extract only the required fields or only the
+        non-required ones.
+    """
+    extracted_fields = {}
+    for key, value in fields.items():
+        if issubclass(type(value), FieldInfo):
+            if required and value.default in [PydanticUndefined, Ellipsis]:
+                extracted_fields[key] = None
+            elif not (required or value.default in [PydanticUndefined, Ellipsis]):
+                if is_jsonable(value.default):
+                    extracted_fields[key] = value.default
+                elif issubclass_patched(value.default, BaseModel):
+                    extracted_fields[key] = value.default.model_dump(mode="json")
+                else:
+                    extracted_fields[key] = str(value.default)
+        else:
+            extracted_fields[key] = extract_config_fields_for_yaml(
+                fields[key], required
+            )
+    return extracted_fields
+
+
+def create_config(file_name: str, dir_path: Path, include_optional: bool) -> None:
+    """Create a KPOps config yaml.
+
+    :param file_name: Name for the file
+    :param dir_path: Directory in which the file should be created
+    :param include_optional: Whether to include non-required settings
+    """
+    file_path = Path(dir_path / (file_name + ".yaml"))
+    file_path.touch(exist_ok=False)
+    with file_path.open(mode="w") as conf:
+        conf.write("# " + describe_object(KpopsConfig.__doc__))  # Write title
+        non_required = extract_config_fields_for_yaml(
+            collect_fields(KpopsConfig), False
+        )
+        required = extract_config_fields_for_yaml(collect_fields(KpopsConfig), True)
+        for k in non_required:
+            required.pop(k, None)
+        conf.write("\n\n# Required fields\n")
+        conf.write(yaml.dump(required))
+        if include_optional:
+            conf.write("\n# Non-required fields\n")
+            conf.write(yaml.dump(non_required))
+
+
+def init_project(path: Path, conf_incl_opt: bool):
+    """Initiate a default empty project.
+
+    :param path: Directory in which the project should be initiated
+    :param conf_incl_opt: Whether to include non-required settings
+        in the generated config file
+    """
+    create_config("config", path, conf_incl_opt)
+    Path(path / ("pipeline.yaml")).touch(exist_ok=False)
+    Path(path / ("defaults.yaml")).touch(exist_ok=False)

kpops/utils/docstring.py

@@ -59,6 +59,7 @@ def _trim_description_end(desc: str) -> str:
     desc_enders = [
         ":param ",
         ":returns:",
+        ":raises:",
         "defaults to ",
     ]
     end_index = len(desc)

kpops/utils/json.py

@@ -0,0 +1,15 @@
+import json
+from typing import Any
+
+
+def is_jsonable(input: Any) -> bool:
+    """Check whether a value is json-serializable.
+
+    :param input: Value to be checked.
+    """
+    try:
+        json.dumps(input)
+    except (TypeError, OverflowError):
+        return False
+    else:
+        return True

kpops/utils/pydantic.py

@@ -109,8 +109,8 @@ def issubclass_patched(
         issubclass(BaseSettings, BaseModel)  # True
         issubclass(set[str], BaseModel)  # raises Exception
 
-    :param cls: class to check
-    :base: class(es) to check against, defaults to ``BaseModel``
+    :param __cls: class to check
+    :param __class_or_tuple: class(es) to check against, defaults to ``BaseModel``
     :return: Whether 'cls' is derived from another class or is the same class.
     """
     try:

kpops/utils/yaml.py

@@ -15,6 +15,17 @@
 log = logging.getLogger("Yaml")
 
 
+def touch_yaml_file(file_name, dir_path) -> Path:
+    """Create an empty YAML file.
+
+    :param file_name: Name of the new file
+    :param dir_path: Parent directory of the new file
+    """
+    file_path = Path(dir_path / (file_name + ".yaml"))
+    file_path.touch(exist_ok=False)
+    return file_path
+
+
 def generate_hashkey(
     file_path: Path, substitution: Mapping[str, Any] | None = None
 ) -> tuple:

tests/utils/test_cli_commands.py

@@ -0,0 +1,40 @@
+from pathlib import Path
+
+from typer.testing import CliRunner
+
+import kpops
+from kpops.cli.main import app
+from kpops.utils.cli_commands import create_config
+
+runner = CliRunner()
+
+
+def test_create_config(tmp_path: Path):
+    opt_conf_name = "config_with_non_required"
+    req_conf_name = "config_with_only_required"
+    create_config(opt_conf_name, tmp_path, True)
+    create_config(req_conf_name, tmp_path, False)
+    assert (opt_conf := Path(tmp_path / (opt_conf_name + ".yaml"))).exists()
+    assert (req_conf := Path(tmp_path / (req_conf_name + ".yaml"))).exists()
+    with opt_conf.open() as opt_file, req_conf.open() as req_file:
+        assert len(opt_file.readlines()) > len(req_file.readlines())
+
+
+def test_init_project(tmp_path: Path):
+    kpops.init(tmp_path, config_include_opt=True)
+    for path in ["config.yaml", "defaults.yaml", "pipeline.yaml"]:
+        assert Path(tmp_path / path).exists()
+
+
+def test_init_project_from_cli_with_bad_path(tmp_path: Path):
+    bad_path = Path(tmp_path / "random_file.yaml")
+    bad_path.touch()
+    result = runner.invoke(
+        app,
+        [
+            "init",
+            str(bad_path),
+        ],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 2