[components] Factor out click error message code

python_modules/libraries/dagster-dg/dagster_dg/cli/__init__.py

@@ -1,4 +1,3 @@
-import sys
 from pathlib import Path
 
 import click
@@ -11,7 +10,7 @@
 from dagster_dg.component import RemoteComponentRegistry
 from dagster_dg.config import normalize_cli_config
 from dagster_dg.context import DgContext
-from dagster_dg.utils import DgClickGroup
+from dagster_dg.utils import DgClickGroup, exit_with_error
 from dagster_dg.version import __version__
 
 DG_CLI_MAX_OUTPUT_WIDTH = 120
@@ -59,12 +58,7 @@ def group(
     ):
         """CLI for working with Dagster components."""
         if clear_cache and rebuild_component_registry:
-            click.echo(
-                click.style(
-                    "Cannot specify both --clear-cache and --rebuild-component-registry.", fg="red"
-                )
-            )
-            sys.exit(1)
+            exit_with_error("Cannot specify both --clear-cache and --rebuild-component-registry.")
         elif clear_cache:
             cli_config = normalize_cli_config(global_options, context)
             dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
@@ -75,12 +69,7 @@ def group(
             cli_config = normalize_cli_config(global_options, context)
             dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
             if context.invoked_subcommand is not None:
-                click.echo(
-                    click.style(
-                        "Cannot specify --rebuild-component-registry with a subcommand.", fg="red"
-                    )
-                )
-                sys.exit(1)
+                exit_with_error("Cannot specify --rebuild-component-registry with a subcommand.")
             _rebuild_component_registry(dg_context)
         elif context.invoked_subcommand is None:
             click.echo(context.get_help())
@@ -91,25 +80,13 @@ def group(
 
 def _rebuild_component_registry(dg_context: DgContext):
     if not dg_context.is_code_location:
-        click.echo(
-            click.style(
-                "This command must be run inside a Dagster code location directory.", fg="red"
-            )
-        )
-        sys.exit(1)
+        exit_with_error("This command must be run inside a Dagster code location directory.")
     elif not dg_context.has_cache:
-        click.echo(
-            click.style("Cache is disabled. This command cannot be run without a cache.", fg="red")
-        )
-        sys.exit(1)
+        exit_with_error("Cache is disabled. This command cannot be run without a cache.")
     elif not dg_context.config.use_dg_managed_environment:
-        click.echo(
-            click.style(
-                "Cannot rebuild the component registry with environment management disabled.",
-                fg="red",
-            )
+        exit_with_error(
+            "Cannot rebuild the component registry with environment management disabled."
         )
-        sys.exit(1)
     dg_context.ensure_uv_lock()
     key = dg_context.get_cache_key("component_registry_data")
     dg_context.cache.clear_key(key)

python_modules/libraries/dagster-dg/dagster_dg/cli/code_location.py

@@ -1,5 +1,4 @@
 import os
-import sys
 from pathlib import Path
 from typing import Optional
 
@@ -9,7 +8,7 @@
 from dagster_dg.config import normalize_cli_config
 from dagster_dg.context import DgContext
 from dagster_dg.scaffold import scaffold_code_location
-from dagster_dg.utils import DgClickCommand, DgClickGroup
+from dagster_dg.utils import DgClickCommand, DgClickGroup, exit_with_error
 
 
 @click.group(name="code-location", cls=DgClickGroup)
@@ -79,21 +78,16 @@ def code_location_scaffold_command(
     dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
     if dg_context.is_deployment:
         if dg_context.has_code_location(name):
-            click.echo(click.style(f"A code location named {name} already exists.", fg="red"))
-            sys.exit(1)
+            exit_with_error(f"A code location named {name} already exists.")
         code_location_path = dg_context.code_location_root_path / name
     else:
         code_location_path = Path.cwd() / name
 
     if use_editable_dagster == "TRUE":
         if not os.environ.get("DAGSTER_GIT_REPO_DIR"):
-            click.echo(
-                click.style(
-                    "The `--use-editable-dagster` flag requires the `DAGSTER_GIT_REPO_DIR` environment variable to be set.",
-                    fg="red",
-                )
+            exit_with_error(
+                "The `--use-editable-dagster` flag requires the `DAGSTER_GIT_REPO_DIR` environment variable to be set."
             )
-            sys.exit(1)
         editable_dagster_root = os.environ["DAGSTER_GIT_REPO_DIR"]
     elif use_editable_dagster:  # a string value was passed
         editable_dagster_root = use_editable_dagster
@@ -118,10 +112,7 @@ def code_location_list_command(context: click.Context, **global_options: object)
     cli_config = normalize_cli_config(global_options, context)
     dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
     if not dg_context.is_deployment:
-        click.echo(
-            click.style("This command must be run inside a Dagster deployment directory.", fg="red")
-        )
-        sys.exit(1)
+        exit_with_error("This command must be run inside a Dagster deployment directory.")
 
     for code_location in dg_context.get_code_location_names():
         click.echo(code_location)

python_modules/libraries/dagster-dg/dagster_dg/cli/component.py

@@ -20,6 +20,7 @@
 from dagster_dg.utils import (
     DgClickCommand,
     DgClickGroup,
+    exit_with_error,
     json_schema_property_to_click_option,
     not_none,
     parse_json_option,
@@ -67,12 +68,7 @@ def _define_commands(self, cli_context: click.Context) -> None:
         dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), config)
 
         if not dg_context.is_code_location:
-            click.echo(
-                click.style(
-                    "This command must be run inside a Dagster code location directory.", fg="red"
-                )
-            )
-            sys.exit(1)
+            exit_with_error("This command must be run inside a Dagster code location directory.")
 
         registry = RemoteComponentRegistry.from_dg_context(dg_context)
         for key, component_type in registry.items():
@@ -181,26 +177,13 @@ def scaffold_component_command(
         cli_config = get_config_from_cli_context(cli_context)
         dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
         if not dg_context.is_code_location:
-            click.echo(
-                click.style(
-                    "This command must be run inside a Dagster code location directory.", fg="red"
-                )
-            )
-            sys.exit(1)
+            exit_with_error("This command must be run inside a Dagster code location directory.")
 
         registry = RemoteComponentRegistry.from_dg_context(dg_context)
         if not registry.has(component_key):
-            click.echo(
-                click.style(f"No component type `{component_key}` could be resolved.", fg="red")
-            )
-            sys.exit(1)
+            exit_with_error(f"No component type `{component_key}` could be resolved.")
         elif dg_context.has_component(component_name):
-            click.echo(
-                click.style(
-                    f"A component instance named `{component_name}` already exists.", fg="red"
-                )
-            )
-            sys.exit(1)
+            exit_with_error(f"A component instance named `{component_name}` already exists.")
 
         # Specified key-value params will be passed to this function with their default value of
         # `None` even if the user did not set them. Filter down to just the ones that were set by
@@ -211,14 +194,10 @@ def scaffold_component_command(
             if cli_context.get_parameter_source(k) == ParameterSource.COMMANDLINE
         }
         if json_params is not None and user_provided_key_value_params:
-            click.echo(
-                click.style(
-                    "Detected params passed as both --json-params and individual options. These are mutually exclusive means of passing"
-                    " component generation parameters. Use only one.",
-                    fg="red",
-                )
+            exit_with_error(
+                "Detected params passed as both --json-params and individual options. These are mutually exclusive means of passing"
+                " component generation parameters. Use only one.",
             )
-            sys.exit(1)
         elif json_params:
             scaffold_params = json_params
         elif user_provided_key_value_params:
@@ -259,12 +238,7 @@ def component_list_command(context: click.Context, **global_options: object) ->
     cli_config = normalize_cli_config(global_options, context)
     dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
     if not dg_context.is_code_location:
-        click.echo(
-            click.style(
-                "This command must be run inside a Dagster code location directory.", fg="red"
-            )
-        )
-        sys.exit(1)
+        exit_with_error("This command must be run inside a Dagster code location directory.")
 
     for component_name in dg_context.get_component_names():
         click.echo(component_name)

python_modules/libraries/dagster-dg/dagster_dg/cli/component_type.py

@@ -1,5 +1,4 @@
 import json
-import sys
 from collections.abc import Mapping
 from pathlib import Path
 from typing import Any
@@ -12,7 +11,7 @@
 from dagster_dg.context import DgContext
 from dagster_dg.docs import markdown_for_component_type, render_markdown_in_browser
 from dagster_dg.scaffold import scaffold_component_type
-from dagster_dg.utils import DgClickCommand, DgClickGroup
+from dagster_dg.utils import DgClickCommand, DgClickGroup, exit_with_error
 
 
 @click.group(name="component-type", cls=DgClickGroup)
@@ -40,17 +39,11 @@ def component_type_scaffold_command(
     cli_config = normalize_cli_config(global_options, context)
     dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
     if not dg_context.is_code_location:
-        click.echo(
-            click.style(
-                "This command must be run inside a Dagster code location directory.", fg="red"
-            )
-        )
-        sys.exit(1)
+        exit_with_error("This command must be run inside a Dagster code location directory.")
     registry = RemoteComponentRegistry.from_dg_context(dg_context)
     full_component_name = f"{dg_context.root_package_name}.{name}"
     if registry.has(full_component_name):
-        click.echo(click.style(f"A component type named `{name}` already exists.", fg="red"))
-        sys.exit(1)
+        exit_with_error(f"A component type named `{name}` already exists.")
 
     scaffold_component_type(dg_context, name)
 
@@ -74,10 +67,7 @@ def component_type_docs_command(
     dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
     registry = RemoteComponentRegistry.from_dg_context(dg_context)
     if not registry.has(component_type):
-        click.echo(
-            click.style(f"No component type `{component_type}` could be resolved.", fg="red")
-        )
-        sys.exit(1)
+        exit_with_error(f"No component type `{component_type}` could be resolved.")
 
     render_markdown_in_browser(markdown_for_component_type(registry.get(component_type)))
 
@@ -107,18 +97,11 @@ def component_type_info_command(
     dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
     registry = RemoteComponentRegistry.from_dg_context(dg_context)
     if not registry.has(component_type):
-        click.echo(
-            click.style(f"No component type `{component_type}` could be resolved.", fg="red")
-        )
-        sys.exit(1)
+        exit_with_error(f"No component type `{component_type}` could be resolved.")
     elif sum([description, scaffold_params_schema, component_params_schema]) > 1:
-        click.echo(
-            click.style(
-                "Only one of --description, --scaffold-params-schema, and --component-params-schema can be specified.",
-                fg="red",
-            )
+        exit_with_error(
+            "Only one of --description, --scaffold-params-schema, and --component-params-schema can be specified."
         )
-        sys.exit(1)
 
     component_type_metadata = registry.get(component_type)
 

python_modules/libraries/dagster-dg/dagster_dg/cli/deployment.py

@@ -8,7 +8,7 @@
 from dagster_dg.config import normalize_cli_config
 from dagster_dg.context import DgContext
 from dagster_dg.scaffold import scaffold_deployment
-from dagster_dg.utils import DgClickCommand, DgClickGroup
+from dagster_dg.utils import DgClickCommand, DgClickGroup, exit_with_error
 
 
 @click.group(name="deployment", cls=DgClickGroup)
@@ -37,8 +37,8 @@ def deployment_scaffold_command(
     dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
     dir_abspath = os.path.abspath(path)
     if os.path.exists(dir_abspath):
-        click.echo(
-            click.style(f"A file or directory at {dir_abspath} already exists. ", fg="red")
+        exit_with_error(
+            f"A file or directory at {dir_abspath} already exists. "
             + "\nPlease delete the contents of this path or choose another location."
         )
         sys.exit(1)

python_modules/libraries/dagster-dg/dagster_dg/utils.py

@@ -239,6 +239,11 @@ def not_none(value: Optional[T]) -> T:
     return value
 
 
+def exit_with_error(error_message: str) -> None:
+    click.echo(click.style(error_message, fg="red"))
+    sys.exit(1)
+
+
 # ########################
 # ##### CUSTOM CLICK SUBCLASSES
 # ########################