add universal exception wrapper and configurable docs urls

dlt/cli/_dlt.py

@@ -1,11 +1,13 @@
 from typing import Any, Sequence, Type, cast, List, Dict
 import argparse
+import click
 
 from dlt.version import __version__
 from dlt.common.runners import Venv
 from dlt.cli import SupportsCliCommand
 
 import dlt.cli.echo as fmt
+from dlt.cli.exceptions import CliCommandException
 
 from dlt.cli.command_wrappers import (
     deploy_command_wrapper,
@@ -15,6 +17,7 @@
 
 
 ACTION_EXECUTED = False
+DEFAULT_DOCS_URL = "https://dlthub.com/docs/intro"
 
 
 def print_help(parser: argparse.ArgumentParser) -> None:
@@ -153,12 +156,35 @@ def main() -> int:
             " the current virtual environment instead."
         )
 
-    if args.command in installed_commands:
-        return installed_commands[args.command].execute(args)
+    if cmd := installed_commands.get(args.command):
+        try:
+            cmd.execute(args)
+        except Exception as ex:
+            docs_url = cmd.docs_url if hasattr(cmd, "docs_url") else DEFAULT_DOCS_URL
+            error_code = -1
+            raiseable_exception = ex
+
+            # overwrite some values if this is a CliCommandException
+            if isinstance(ex, CliCommandException):
+                error_code = ex.error_code
+                docs_url = ex.docs_url or docs_url
+                raiseable_exception = ex.raiseable_exception
+
+            # print exception if available
+            if raiseable_exception:
+                click.secho(str(ex), err=True, fg="red")
+
+            fmt.note(f"Please refer to our docs at '%s' for further assistance." % docs_url)
+            if debug.is_debug_enabled() and raiseable_exception:
+                raise raiseable_exception
+
+            return error_code
     else:
         print_help(parser)
         return -1
 
+    return 0
+
 
 def _main() -> None:
     """Script entry point"""

dlt/cli/command_wrappers.py

@@ -11,6 +11,7 @@
 import dlt.cli.echo as fmt
 from dlt.cli import utils
 from dlt.pipeline.exceptions import CannotRestorePipelineException
+from dlt.cli.exceptions import CliCommandException
 
 from dlt.cli.init_command import (
     init_command,
@@ -35,62 +36,41 @@
     pass
 
 
-def on_exception(ex: Exception, info: str) -> None:
-    click.secho(str(ex), err=True, fg="red")
-    fmt.note("Please refer to %s for further assistance" % fmt.bold(info))
-    if debug.is_debug_enabled():
-        raise ex
-
-
 @utils.track_command("init", False, "source_name", "destination_type")
 def init_command_wrapper(
     source_name: str,
     destination_type: str,
     repo_location: str,
     branch: str,
     omit_core_sources: bool = False,
-) -> int:
-    try:
-        init_command(
-            source_name,
-            destination_type,
-            repo_location,
-            branch,
-            omit_core_sources,
-        )
-    except Exception as ex:
-        on_exception(ex, DLT_INIT_DOCS_URL)
-        return -1
-    return 0
+) -> None:
+    init_command(
+        source_name,
+        destination_type,
+        repo_location,
+        branch,
+        omit_core_sources,
+    )
 
 
 @utils.track_command("list_sources", False)
-def list_sources_command_wrapper(repo_location: str, branch: str) -> int:
-    try:
-        list_sources_command(repo_location, branch)
-    except Exception as ex:
-        on_exception(ex, DLT_INIT_DOCS_URL)
-        return -1
-    return 0
+def list_sources_command_wrapper(repo_location: str, branch: str) -> None:
+    list_sources_command(repo_location, branch)
 
 
 @utils.track_command("pipeline", True, "operation")
 def pipeline_command_wrapper(
     operation: str, pipeline_name: str, pipelines_dir: str, verbosity: int, **command_kwargs: Any
-) -> int:
+) -> None:
     try:
         pipeline_command(operation, pipeline_name, pipelines_dir, verbosity, **command_kwargs)
-        return 0
     except CannotRestorePipelineException as ex:
         click.secho(str(ex), err=True, fg="red")
         click.secho(
             "Try command %s to restore the pipeline state from destination"
             % fmt.bold(f"dlt pipeline {pipeline_name} sync")
         )
-        return -1
-    except Exception as ex:
-        on_exception(ex, DLT_PIPELINE_COMMAND_DOCS_URL)
-        return -2
+        raise CliCommandException(error_code=-2)
 
 
 @utils.track_command("deploy", False, "deployment_method")
@@ -100,13 +80,12 @@ def deploy_command_wrapper(
     repo_location: str,
     branch: Optional[str] = None,
     **kwargs: Any,
-) -> int:
+) -> None:
     try:
         utils.ensure_git_command("deploy")
     except Exception as ex:
         click.secho(str(ex), err=True, fg="red")
-        return -1
-
+        raise CliCommandException(error_code=-2)
     from git import InvalidGitRepositoryError, NoSuchPathError
 
     try:
@@ -121,8 +100,7 @@ def deploy_command_wrapper(
         fmt.note(
             "You must run the pipeline locally successfully at least once in order to deploy it."
         )
-        on_exception(ex, DLT_DEPLOY_DOCS_URL)
-        return -2
+        raise CliCommandException(error_code=-3, raiseable_exception=ex)
     except InvalidGitRepositoryError:
         click.secho(
             "No git repository found for pipeline script %s." % fmt.bold(pipeline_script_path),
@@ -140,18 +118,14 @@ def deploy_command_wrapper(
             )
         )
         fmt.note("Please refer to %s for further assistance" % fmt.bold(DLT_DEPLOY_DOCS_URL))
-        return -3
+        raise CliCommandException(error_code=-4)
     except NoSuchPathError as path_ex:
         click.secho("The pipeline script does not exist\n%s" % str(path_ex), err=True, fg="red")
-        return -4
-    except Exception as ex:
-        on_exception(ex, DLT_DEPLOY_DOCS_URL)
-        return -5
-    return 0
+        raise CliCommandException(error_code=-5)
 
 
 @utils.track_command("schema", False, "operation")
-def schema_command_wrapper(file_path: str, format_: str, remove_defaults: bool) -> int:
+def schema_command_wrapper(file_path: str, format_: str, remove_defaults: bool) -> None:
     with open(file_path, "rb") as f:
         if os.path.splitext(file_path)[1][1:] == "json":
             schema_dict: DictStrAny = json.load(f)
@@ -163,24 +137,16 @@ def schema_command_wrapper(file_path: str, format_: str, remove_defaults: bool)
     else:
         schema_str = s.to_pretty_yaml(remove_defaults=remove_defaults)
     fmt.echo(schema_str)
-    return 0
 
 
 @utils.track_command("telemetry", False)
-def telemetry_status_command_wrapper() -> int:
-    try:
-        telemetry_status_command()
-    except Exception as ex:
-        on_exception(ex, DLT_TELEMETRY_DOCS_URL)
-        return -1
-    return 0
+def telemetry_status_command_wrapper() -> None:
+    telemetry_status_command()
 
 
 @utils.track_command("telemetry_switch", False, "enabled")
-def telemetry_change_status_command_wrapper(enabled: bool) -> int:
+def telemetry_change_status_command_wrapper(enabled: bool) -> None:
     try:
         change_telemetry_status_command(enabled)
     except Exception as ex:
-        on_exception(ex, DLT_TELEMETRY_DOCS_URL)
-        return -1
-    return 0
+        raise CliCommandException(docs_url=DLT_TELEMETRY_DOCS_URL, raiseable_exception=ex)

dlt/cli/deploy_command_helpers.py

@@ -35,7 +35,7 @@
 
 from dlt.cli import utils
 from dlt.cli import echo as fmt
-from dlt.cli.exceptions import CliCommandException
+from dlt.cli.exceptions import CliCommandInnerException
 
 GITHUB_URL = "https://github.com/"
 
@@ -99,14 +99,14 @@ def _get_origin(self) -> str:
         try:
             origin = get_origin(self.repo)
             if "github.com" not in origin:
-                raise CliCommandException(
+                raise CliCommandInnerException(
                     "deploy",
                     f"Your current repository origin is not set to github but to {origin}.\nYou"
                     " must change it to be able to run the pipelines with github actions:"
                     " https://docs.github.com/en/get-started/getting-started-with-git/managing-remote-repositories",
                 )
         except ValueError:
-            raise CliCommandException(
+            raise CliCommandInnerException(
                 "deploy",
                 "Your current repository has no origin set. Please set it up to be able to run the"
                 " pipelines with github actions:"
@@ -293,7 +293,7 @@ def get_state_and_trace(pipeline: Pipeline) -> Tuple[TPipelineState, PipelineTra
 def get_visitors(pipeline_script: str, pipeline_script_path: str) -> PipelineScriptVisitor:
     visitor = utils.parse_init_script("deploy", pipeline_script, pipeline_script_path)
     if n.RUN not in visitor.known_calls:
-        raise CliCommandException(
+        raise CliCommandInnerException(
             "deploy",
             f"The pipeline script {pipeline_script_path} does not seem to run the pipeline.",
         )
@@ -323,13 +323,13 @@ def parse_pipeline_info(visitor: PipelineScriptVisitor) -> List[Tuple[str, Optio
                         " abort to set it to False?",
                         default=True,
                     ):
-                        raise CliCommandException("deploy", "Please set the dev_mode to False")
+                        raise CliCommandInnerException("deploy", "Please set the dev_mode to False")
 
             p_d_node = call_args.arguments.get("pipelines_dir")
             if p_d_node:
                 pipelines_dir = evaluate_node_literal(p_d_node)
                 if pipelines_dir is None:
-                    raise CliCommandException(
+                    raise CliCommandInnerException(
                         "deploy",
                         "The value of 'pipelines_dir' argument in call to `dlt_pipeline` cannot be"
                         f" determined from {unparse(p_d_node).strip()}. Pipeline working dir will"
@@ -340,7 +340,7 @@ def parse_pipeline_info(visitor: PipelineScriptVisitor) -> List[Tuple[str, Optio
             if p_n_node:
                 pipeline_name = evaluate_node_literal(p_n_node)
                 if pipeline_name is None:
-                    raise CliCommandException(
+                    raise CliCommandInnerException(
                         "deploy",
                         "The value of 'pipeline_name' argument in call to `dlt_pipeline` cannot be"
                         f" determined from {unparse(p_d_node).strip()}. Pipeline working dir will"
@@ -439,9 +439,9 @@ def ask_files_overwrite(files: Sequence[str]) -> None:
     if existing:
         fmt.echo("Following files will be overwritten: %s" % fmt.bold(str(existing)))
         if not fmt.confirm("Do you want to continue?", default=False):
-            raise CliCommandException("init", "Aborted")
+            raise CliCommandInnerException("init", "Aborted")
 
 
-class PipelineWasNotRun(CliCommandException):
+class PipelineWasNotRun(CliCommandInnerException):
     def __init__(self, msg: str) -> None:
         super().__init__("deploy", msg, None)

dlt/cli/exceptions.py

@@ -1,13 +1,27 @@
 from dlt.common.exceptions import DltException
 
 
-class CliCommandException(DltException):
+class CliCommandInnerException(DltException):
     def __init__(self, cmd: str, msg: str, inner_exc: Exception = None) -> None:
         self.cmd = cmd
         self.inner_exc = inner_exc
         super().__init__(msg)
 
 
+class CliCommandException(DltException):
+    """
+    Exception that can be thrown inside a cli command and can change the
+    error code or docs url presented to the user. Will always be caught.
+    """
+
+    def __init__(
+        self, error_code: int = -1, docs_url: str = None, raiseable_exception: Exception = None
+    ) -> None:
+        self.error_code = error_code
+        self.docs_url = docs_url
+        self.raiseable_exception = raiseable_exception
+
+
 class VerifiedSourceRepoError(DltException):
     def __init__(self, msg: str, source_name: str) -> None:
         self.source_name = source_name

dlt/cli/init_command.py

@@ -36,7 +36,7 @@
     TVerifiedSourceFileEntry,
     TVerifiedSourceFileIndex,
 )
-from dlt.cli.exceptions import CliCommandException
+from dlt.cli.exceptions import CliCommandInnerException
 
 
 DLT_INIT_DOCS_URL = "https://dlthub.com/docs/reference/command-line-interface#dlt-init"
@@ -428,14 +428,14 @@ def init_command(
         source_configuration.src_pipeline_script,
     )
     if visitor.is_destination_imported:
-        raise CliCommandException(
+        raise CliCommandInnerException(
             "init",
             f"The pipeline script {source_configuration.src_pipeline_script} imports a destination"
             " from dlt.destinations. You should specify destinations by name when calling"
             " dlt.pipeline or dlt.run in init scripts.",
         )
     if n.PIPELINE not in visitor.known_calls:
-        raise CliCommandException(
+        raise CliCommandInnerException(
             "init",
             f"The pipeline script {source_configuration.src_pipeline_script} does not seem to"
             " initialize a pipeline with dlt.pipeline. Please initialize pipeline explicitly in"
@@ -498,7 +498,7 @@ def init_command(
             (known_sections.SOURCES, source_name),
         )
         if len(checked_sources) == 0:
-            raise CliCommandException(
+            raise CliCommandInnerException(
                 "init",
                 f"The pipeline script {source_configuration.src_pipeline_script} is not creating or"
                 " importing any sources or resources. Exiting...",
@@ -552,7 +552,7 @@ def init_command(
             )
 
         if not fmt.confirm("Do you want to proceed?", default=True):
-            raise CliCommandException("init", "Aborted")
+            raise CliCommandInnerException("init", "Aborted")
 
     dependency_system = _get_dependency_system(dest_storage)
     _welcome_message(

dlt/cli/pipeline_command.py

@@ -1,7 +1,7 @@
 import yaml
 from typing import Any, Optional, Sequence, Tuple
 import dlt
-from dlt.cli.exceptions import CliCommandException
+from dlt.cli.exceptions import CliCommandInnerException
 
 from dlt.common.json import json
 from dlt.common.pipeline import resource_state, get_dlt_pipelines_dir, TSourceState
@@ -21,7 +21,9 @@
 from dlt.cli import echo as fmt
 
 
-DLT_PIPELINE_COMMAND_DOCS_URL = "https://dlthub.com/docs/reference/command-line-interface"
+DLT_PIPELINE_COMMAND_DOCS_URL = (
+    "https://dlthub.com/docs/reference/command-line-interface#dlt-pipeline"
+)
 
 
 def pipeline_command(
@@ -294,7 +296,7 @@ def _display_pending_packages() -> Tuple[Sequence[str], Sequence[str]]:
             if not packages:
                 packages = sorted(p.list_completed_load_packages())
             if not packages:
-                raise CliCommandException(
+                raise CliCommandInnerException(
                     "pipeline", "There are no load packages for that pipeline"
                 )
             load_id = packages[-1]