Clean the ifs, update the echo command

dlt/cli/echo.py

@@ -47,9 +47,11 @@ def note(msg: str) -> None:
 
 def confirm(text: str, default: Optional[bool] = None) -> bool:
     if ALWAYS_CHOOSE_VALUE:
+        warning("Automatically choosing {ALWAYS_CHOOSE_VALUE} for: {text}")
         return bool(ALWAYS_CHOOSE_VALUE)
     if ALWAYS_CHOOSE_DEFAULT:
-        assert default is not None
+        assert default is not None, "Default value must be provided when ALWAYS_CHOOSE_DEFAULT is enabled."
+        warning(f"Automatically choosing default ({default}) value for: {text}")
         return default
     return click.confirm(text, default=default)
 

dlt/cli/init_command.py

@@ -240,13 +240,9 @@ def init_command(source_name: str, destination_name: str, use_generic_template:
         msg = f"This pipeline requires a newer version of dlt than your installed version ({source_files.requirements.current_dlt_version()}). " \
             f"Pipeline requires '{source_files.requirements.dlt_requirement_base}'"
         fmt.warning(msg)
-        if is_notebook():
-            fmt.warning("Automatically proceeding with init in notebook mode.")
-            pass
-        else:
-            if not fmt.confirm("Would you like to continue anyway? (you can update dlt after this step)", default=True):
-                fmt.echo(f'You can update dlt with: pip3 install -U "{source_files.requirements.dlt_requirement_base}"')
-                return
+        if not fmt.confirm("Would you like to continue anyway? (you can update dlt after this step)", default=True):
+            fmt.echo(f'You can update dlt with: pip3 install -U "{source_files.requirements.dlt_requirement_base}"')
+            return
 
     # read module source and parse it
     visitor = utils.parse_init_script("init", source_files.storage.load(source_files.pipeline_script), source_files.pipeline_script)
@@ -315,11 +311,9 @@ def init_command(source_name: str, destination_name: str, use_generic_template:
             fmt.echo("Cloning and configuring a verified source %s (%s)" % (fmt.bold(source_name), source_files.doc))
             if use_generic_template:
                 fmt.warning("--generic parameter is meaningless if verified source is found")
-        if is_notebook():
-            fmt.warning("Automatically proceeding with init in notebook mode.")
-        else:
-            if not fmt.confirm("Do you want to proceed?", default=True):
-                raise CliCommandException("init", "Aborted")
+
+        if not fmt.confirm("Do you want to proceed?", default=True):
+            raise CliCommandException("init", "Aborted")
 
     dependency_system = _get_dependency_system(dest_storage)
     _welcome_message(source_name, destination_name, source_files, dependency_system, is_new_source)

dlt/cli/magics.py

@@ -12,8 +12,7 @@
 
 
 from ..common.runtime.exec_info import is_notebook, is_ipython
-from dlt.cli import echo as fmt
-
+from dlt.cli import echo as fmt, echo
 
 import typing as t  # type: ignore
 from hyperscript import h  # type: ignore
@@ -128,18 +127,18 @@ def init(self, line:str) -> t.Any:
         args = parse_argstring(self.init, line)
         try:
             from dlt.cli._dlt import init_command_wrapper
-
-            out = init_command_wrapper(
-                source_name=args.source_name,
-                destination_name=args.destination_name,
-                use_generic_template=args.use_generic_template,
-                repo_location=args.repo_location if args.repo_location is not None else DEFAULT_VERIFIED_SOURCES_REPO,
-                branch=args.branch if args.branch is not None else None
-            )
-            if out == -1:
-                return self.display(self.on_exception("Failure due to...", "Default value for init is 'No' for safety reasons."))
-            else:
-                return self.display(self.success_message({"green-bold": "DLT project initialized successfully."}))
+            with echo.always_choose(False, always_choose_value=True):
+                out = init_command_wrapper(
+                    source_name=args.source_name,
+                    destination_name=args.destination_name,
+                    use_generic_template=args.use_generic_template,
+                    repo_location=args.repo_location if args.repo_location is not None else DEFAULT_VERIFIED_SOURCES_REPO,
+                    branch=args.branch if args.branch is not None else None
+                )
+                if out == -1:
+                    return self.display(self.on_exception("Failure due to...", "Default value for init is 'No' for safety reasons."))
+                else:
+                    return self.display(self.success_message({"green-bold": "DLT project initialized successfully."}))
         except Exception as ex:
             self.on_exception(str(ex), DLT_INIT_DOCS_URL)
             return -1
@@ -215,13 +214,14 @@ def pipeline(self, line:str)->t.Any:
             args.operation = 'list'
         try:
             from dlt.cli._dlt import pipeline_command_wrapper, DLT_PIPELINE_COMMAND_DOCS_URL
-            pipeline_command_wrapper(
-                operation=args.operation,
-                pipeline_name=args.pipeline_name,
-                pipelines_dir=args.pipelines_dir,
-                verbosity=args.verbosity
-
-            )
+            with echo.always_choose(False, always_choose_value=True):
+                pipeline_command_wrapper(
+                    operation=args.operation,
+                    pipeline_name=args.pipeline_name,
+                    pipelines_dir=args.pipelines_dir,
+                    verbosity=args.verbosity
+
+                )
         except Exception as ex:
             self.on_exception(str(ex), DLT_PIPELINE_COMMAND_DOCS_URL)
             return -2
@@ -238,12 +238,13 @@ def schema(self, line:str)->t.Any:
         args = parse_argstring(self.schema, line)
         try:
             from dlt.cli._dlt import schema_command_wrapper
-            schema_command_wrapper(
-                file_path=args.file_path,
-                format_=args.format,
-                remove_defaults=args.remove_defaults
-            )
-            return self.display(self.success_message({"green-bold": "DLT schema magic ran successfully."}))
+            with echo.always_choose(False, always_choose_value=True):
+                schema_command_wrapper(
+                    file_path=args.file_path,
+                    format_=args.format,
+                    remove_defaults=args.remove_defaults
+                )
+                return self.display(self.success_message({"green-bold": "DLT schema magic ran successfully."}))
         except Exception as ex:
             self.on_exception(str(ex), "Schema Documentation URL")  # Replace with actual URL
             return -1

dlt/cli/pipeline_command.py

@@ -41,10 +41,6 @@ def pipeline_command(operation: str, pipeline_name: str, pipelines_dir: str, ver
         if operation not in {"sync", "drop"}:
             raise
         fmt.warning(str(e))
-        if is_notebook():
-            fmt.warning("Automatically proceeding without pipeline restoration in notebook mode.")
-            return
-
         if not fmt.confirm("Do you want to attempt to restore the pipeline state from destination?", default=False):
             return
         destination = destination or fmt.text_input(f"Enter destination name for pipeline {fmt.bold(pipeline_name)}")
@@ -167,30 +163,16 @@ def _display_pending_packages() -> Tuple[Sequence[str], Sequence[str]]:
         extracted_files, norm_packages = _display_pending_packages()
         if len(extracted_files) == 0 and len(norm_packages) == 0:
             fmt.echo("No pending packages found")
-        if is_notebook():
-            fmt.warning("Automatically proceeding with package deletion in notebook mode.")
+        if fmt.confirm("Delete the above packages?", default=False):
             p.drop_pending_packages(with_partial_loads=True)
             fmt.echo("Pending packages deleted")
 
-        else:
-            if fmt.confirm("Delete the above packages?", default=False):
-                p.drop_pending_packages(with_partial_loads=True)
-                fmt.echo("Pending packages deleted")
-
     if operation == "sync":
-        if is_notebook():
-            fmt.warning("Automatically proceeding with: drop the local state of the pipeline")
+        if fmt.confirm("About to drop the local state of the pipeline and reset all the schemas. The destination state, data and schemas are left intact. Proceed?", default=False):
             fmt.echo("Dropping local state")
             p = p.drop()
             fmt.echo("Restoring from destination")
             p.sync_destination()
-        else:
-
-            if fmt.confirm("About to drop the local state of the pipeline and reset all the schemas. The destination state, data and schemas are left intact. Proceed?", default=False):
-                fmt.echo("Dropping local state")
-                p = p.drop()
-                fmt.echo("Restoring from destination")
-                p.sync_destination()
 
     if operation == "load-package":
         load_id = command_kwargs.get('load_id')
@@ -242,9 +224,6 @@ def _display_pending_packages() -> Tuple[Sequence[str], Sequence[str]]:
         #     fmt.echo("%s: %s" % (fmt.style(k, fg="green"), v))
         for warning in drop.info["warnings"]:
             fmt.warning(warning)
-        if is_notebook():
-            fmt.warning("Automatically proceeding with dropping the data in notebook mode.")
+
+        if fmt.confirm("Do you want to apply these changes?", default=False):
             drop()
-        else:
-            if fmt.confirm("Do you want to apply these changes?", default=False):
-                drop()

docs/website/docs/reference/notebook-magic.md

@@ -20,22 +20,13 @@ In an IPython notebook, you can use the `%init` magic command to execute the `dl
 
 ```
 
-## `%deploy`
-
-Deploy a DLT pipeline to a destination.
-
-```ipython
-
-%deploy --pipeline_script_path <path> --deployment_method <git-actions> --repo_location <repo location> --branch <branch> --schedule <5 minutes, 15 minutes, 30 minutes, 45 minutes or hour> --run_on_push <> --run_on_schedule <True/False> --run_on_manual <True/False>
-
-```
 
 ## `%pipeline`
 
 In an IPython notebook, you can use the `%init` magic command to execute the `dlt init` command. This command sets up a new DLT pipeline script that transfers data from a `source` to a `destination`.
 
 ```ipython
-%pipeline --operation <operation>  -- <destination>
+%pipeline --operation <operation>  --destination <destination>
 
 ```