Rewrite KFP code generation

docs/source/user_guide/pipelines.md

@@ -372,12 +372,16 @@ To export a pipeline from the Visual Pipeline Editor:
 
 #### Exporting a pipeline from the command line interface
 
-Use the [`elyra-pipeline`](command-line-interface.html#working-with-pipelines) `export` command to export a pipeline to a runtime-specific format, such as YAML for Kubeflow Pipelines or Python DAG for Apache Airflow.
+Use the [`elyra-pipeline`](command-line-interface.html#working-with-pipelines) `export` command to export a pipeline to a runtime-specific format:
+- Kubeflow Pipelines: [Python DSL](https://v1-5-branch.kubeflow.org/docs/components/pipelines/sdk/build-pipeline/) or YAML
+- Apache Airflow: Python DAG
 
 ```bash
 $ elyra-pipeline export a-notebook.pipeline --runtime-config kfp_dev_env --output /path/to/exported.yaml --overwrite
 ```
 
+By default export produces YAML formatted output for Kubeflow Pipelines and Python DAG for Apache Airflow. To choose a different format, specify the `--format` option. Supported values are `py` and `yaml` for Kubeflow Pipelines and `py` for Apache Airflow.
+
 To learn more about supported parameters, run
 ```bash
 $ elyra-pipeline export --help

elyra/cli/pipeline_app.py

@@ -670,10 +670,16 @@ def describe(json_option, pipeline_path):
     "--output",
     required=False,
     type=Path,
-    help="Exported file name (including optional path). Defaults to " " the current directory and the pipeline name.",
+    help="Exported file name (including optional path). Defaults to the current directory and the pipeline name.",
+)
+@click.option(
+    "--format",
+    required=False,
+    type=str,
+    help="File export format.",
 )
 @click.option("--overwrite", is_flag=True, help="Overwrite output file if it already exists.")
-def export(pipeline_path, runtime_config, output, overwrite):
+def export(pipeline_path, runtime_config, output, format, overwrite):
     """
     Export a pipeline to a runtime-specific format
     """
@@ -699,14 +705,20 @@ def export(pipeline_path, runtime_config, output, overwrite):
             param_hint="--runtime-config",
         )
 
+    # Determine which export format(s) the runtime processor supports
     resources = RuntimeTypeResources.get_instance_by_type(RuntimeProcessorType.get_instance_by_name(runtime_type))
     supported_export_formats = resources.get_export_extensions()
     if len(supported_export_formats) == 0:
         raise click.ClickException(f"Runtime type '{runtime_type}' does not support export.")
 
-    # If, in the future, a runtime supports multiple export output formats,
-    # the user can choose one. For now, choose the only option.
-    selected_export_format = supported_export_formats[0]
+    # Verify that the user selected a valid format. If none was specified,
+    # the first from the supported list is selected as default.
+    selected_export_format = (format or supported_export_formats[0]).lower()
+    if selected_export_format not in supported_export_formats:
+        raise click.BadParameter(
+            f"Valid export formats are {supported_export_formats}.",
+            param_hint="--format",
+        )
     selected_export_format_suffix = f".{selected_export_format}"
 
     # generate output file name from the user-provided input