Selectively show the multiple files/modules help text to the commands…

… where it applies (#11689)

Summary:
User reported confusion about why dagster api grpc claimed to support
multiple modules. This PR restricts the multiple=True to commands that
support it (commands that interact with the workspace, as opposed to
commands intended to point at a single code location or a single job)

Test Plan:
Existing CLI coverage validates inputs to the CLI functions at play
Inspect the output of dagster api grpc --help and dagster job laaunch
--help

### Summary & Motivation

### How I Tested These Changes

python_modules/dagster/dagster/_cli/api.py

@@ -14,7 +14,6 @@
 from dagster._cli.workspace.cli_target import (
     get_working_directory_from_kwargs,
     python_origin_target_argument,
-    unwrap_single_code_location_target_cli_arg,
 )
 from dagster._core.definitions.metadata import MetadataEntry
 from dagster._core.errors import DagsterExecutionInterruptedError
@@ -680,18 +679,8 @@ def grpc_command(
         ]
     ):
         # in the gRPC api CLI we never load more than one module or python file at a time
-
-        module_name = (
-            unwrap_single_code_location_target_cli_arg(kwargs, "module_name")
-            if kwargs["module_name"]
-            else None
-        )
-
-        python_file = (
-            unwrap_single_code_location_target_cli_arg(kwargs, "python_file")
-            if kwargs["python_file"]
-            else None
-        )
+        module_name = check.opt_str_elem(kwargs, "module_name")
+        python_file = check.opt_str_elem(kwargs, "python_file")
 
         loadable_target_origin = LoadableTargetOrigin(
             executable_path=sys.executable,

python_modules/dagster/dagster/_cli/dev.py

@@ -28,8 +28,8 @@ def dev_command_options(f):
     return apply_click_params(
         f,
         workspace_option(),
-        python_file_option(),
-        python_module_option(),
+        python_file_option(allow_multiple=True),
+        python_module_option(allow_multiple=True),
     )
 
 

python_modules/dagster/dagster/_cli/workspace/cli_target.py

@@ -286,18 +286,20 @@ def get_workspace_from_kwargs(
         yield workspace_process_context.create_request_context()
 
 
-def python_file_option():
+def python_file_option(allow_multiple):
     return click.option(
         "--python-file",
         "-f",
         # Checks that the path actually exists lower in the stack, where we
         # are better equipped to surface errors
         type=click.Path(exists=False),
-        multiple=True,
+        multiple=allow_multiple,
         help=(
-            "Specify python file or files (flag can be used multiple times) where "
-            "dagster definitions reside as top-level symbols/variables and load each "
-            "file as a code location in the current python environment."
+            "Specify python file "
+            + ("or files (flag can be used multiple times) " if allow_multiple else "")
+            + "where dagster definitions reside as top-level symbols/variables and load "
+            + ("each" if allow_multiple else "the")
+            + " file as a code location in the current python environment."
         ),
         envvar="DAGSTER_PYTHON_FILE",
     )
@@ -313,30 +315,32 @@ def workspace_option():
     )
 
 
-def python_module_option():
+def python_module_option(allow_multiple):
     return click.option(
         "--module-name",
         "-m",
-        multiple=True,
+        multiple=allow_multiple,
         help=(
-            "Specify module or modules (flag can be used multiple times) where "
-            "dagster definitions reside as top-level symbols/variables and load each "
-            "module as a code location in the current python environment."
+            "Specify module "
+            + ("or modules (flag can be used multiple times) " if allow_multiple else "")
+            + "where dagster definitions reside as top-level symbols/variables and load "
+            + ("each" if allow_multiple else "the")
+            + " module as a code location in the current python environment."
         ),
         envvar="DAGSTER_MODULE_NAME",
     )
 
 
-def python_target_click_options():
+def python_target_click_options(allow_multiple_python_targets):
     return [
         click.option(
             "--working-directory",
             "-d",
             help="Specify working directory to use when loading the repository or job",
             envvar="DAGSTER_WORKING_DIRECTORY",
         ),
-        python_file_option(),
-        python_module_option(),
+        python_file_option(allow_multiple=allow_multiple_python_targets),
+        python_module_option(allow_multiple=allow_multiple_python_targets),
         click.option(
             "--package-name",
             help="Specify Python package where repository or job function lives",
@@ -389,14 +393,14 @@ def workspace_target_click_options():
             click.option("--empty-workspace", is_flag=True, help="Allow an empty workspace"),
             workspace_option(),
         ]
-        + python_target_click_options()
+        + python_target_click_options(allow_multiple_python_targets=True)
         + grpc_server_target_click_options()
     )
 
 
 def python_job_target_click_options():
     return (
-        python_target_click_options()
+        python_target_click_options(allow_multiple_python_targets=False)
         + [
             click.option(
                 "--repository",
@@ -466,7 +470,7 @@ def grpc_server_origin_target_argument(f):
 def python_origin_target_argument(f):
     from dagster._cli.job import apply_click_params
 
-    options = python_target_click_options()
+    options = python_target_click_options(allow_multiple_python_targets=False)
     return apply_click_params(f, *options)
 
 
@@ -545,16 +549,8 @@ def get_job_python_origin_from_kwargs(kwargs):
 
 
 def _get_code_pointer_dict_from_kwargs(kwargs: ClickArgMapping) -> Mapping[str, CodePointer]:
-    python_file = (
-        unwrap_single_code_location_target_cli_arg(kwargs, "python_file")
-        if kwargs.get("python_file")
-        else None
-    )
-    module_name = (
-        unwrap_single_code_location_target_cli_arg(kwargs, "module_name")
-        if kwargs.get("module_name")
-        else None
-    )
+    python_file = check.opt_str_elem(kwargs, "python_file")
+    module_name = check.opt_str_elem(kwargs, "module_name")
     package_name = check.opt_str_elem(kwargs, "package_name")
     working_directory = get_working_directory_from_kwargs(kwargs)
     attribute = check.opt_str_elem(kwargs, "attribute")
@@ -605,27 +601,7 @@ def get_working_directory_from_kwargs(kwargs: ClickArgMapping) -> Optional[str]:
     return check.opt_str_elem(kwargs, "working_directory") or os.getcwd()
 
 
-def unwrap_single_code_location_target_cli_arg(kwargs: ClickArgMapping, key: str) -> str:
-    """
-    Dagster CLI tools accept multiple code location targets (e.g. multiple -f and -m instances)
-    but sometimes only one makes sense (e.g. when targeting a single job)
-    Use this function to validate that there is only one value in that tuple and then return the tuple itself.
-
-    key can be module_name or python_file
-    """
-    check.is_tuple(kwargs[key], of_type=str)
-    value_tuple = cast(Tuple[str], kwargs[key])
-    check.invariant(
-        len(value_tuple) == 1,
-        (
-            "Must specify only one code location when executing this command. Multiple {key}"
-            " options given"
-        ),
-    )
-    return value_tuple[0]
-
-
-def get_repository_python_origin_from_kwargs(kwargs: ClickArgMapping) -> RepositoryPythonOrigin:
+def get_repository_python_origin_from_kwargs(kwargs: Mapping[str, str]) -> RepositoryPythonOrigin:
     provided_repo_name = cast(str, kwargs.get("repository"))
 
     if not (kwargs.get("python_file") or kwargs.get("module_name") or kwargs.get("package_name")):
@@ -638,15 +614,15 @@ def get_repository_python_origin_from_kwargs(kwargs: ClickArgMapping) -> Reposit
     if kwargs.get("attribute") and not provided_repo_name:
         if kwargs.get("python_file"):
             _check_cli_arguments_none(kwargs, "module_name", "package_name")
-            python_file = unwrap_single_code_location_target_cli_arg(kwargs, "python_file")
+            python_file = check.str_elem(kwargs, "python_file")
             code_pointer: CodePointer = CodePointer.from_python_file(
                 python_file,
                 check.str_elem(kwargs, "attribute"),
                 get_working_directory_from_kwargs(kwargs),
             )
         elif kwargs.get("module_name"):
             _check_cli_arguments_none(kwargs, "python_file", "package_name")
-            module_name = unwrap_single_code_location_target_cli_arg(kwargs, "module_name")
+            module_name = check.str_elem(kwargs, "module_name")
             code_pointer = CodePointer.from_module(
                 module_name,
                 check.str_elem(kwargs, "attribute"),

python_modules/dagster/dagster_tests/cli_tests/command_tests/test_cli_commands.py

@@ -479,6 +479,16 @@ def grpc_server_backfill_args():
 
 
 def non_existant_python_origin_target_args():
+    return {
+        "workspace": None,
+        "job_name": "foo",
+        "python_file": file_relative_path(__file__, "made_up_file.py"),
+        "module_name": None,
+        "attribute": "bar",
+    }
+
+
+def non_existant_python_file_workspace_args():
     return {
         "workspace": None,
         "job_name": "foo",
@@ -496,14 +506,14 @@ def valid_job_python_origin_target_args():
         {
             "workspace": None,
             "job_name": job_name,
-            "python_file": (file_relative_path(__file__, "test_cli_commands.py"),),
+            "python_file": file_relative_path(__file__, "test_cli_commands.py"),
             "module_name": None,
             "attribute": "bar",
         },
         {
             "workspace": None,
             "job_name": job_name,
-            "python_file": (file_relative_path(__file__, "test_cli_commands.py"),),
+            "python_file": file_relative_path(__file__, "test_cli_commands.py"),
             "module_name": None,
             "attribute": "bar",
             "working_directory": os.path.dirname(__file__),
@@ -512,14 +522,14 @@ def valid_job_python_origin_target_args():
             "workspace": None,
             "job_name": job_name,
             "python_file": None,
-            "module_name": ("dagster_tests.cli_tests.command_tests.test_cli_commands",),
+            "module_name": "dagster_tests.cli_tests.command_tests.test_cli_commands",
             "attribute": "bar",
         },
         {
             "workspace": None,
             "job_name": job_name,
             "python_file": None,
-            "module_name": ("dagster_tests.cli_tests.command_tests.test_cli_commands",),
+            "module_name": "dagster_tests.cli_tests.command_tests.test_cli_commands",
             "attribute": "bar",
             "working_directory": os.path.dirname(__file__),
         },
@@ -542,7 +552,7 @@ def valid_job_python_origin_target_args():
             "workspace": None,
             "job_name": None,
             "python_file": None,
-            "module_name": ("dagster_tests.cli_tests.command_tests.test_cli_commands",),
+            "module_name": "dagster_tests.cli_tests.command_tests.test_cli_commands",
             "attribute": job_fn_name,
         },
         {
@@ -555,28 +565,43 @@ def valid_job_python_origin_target_args():
         {
             "workspace": None,
             "job_name": None,
-            "python_file": (file_relative_path(__file__, "test_cli_commands.py"),),
+            "python_file": file_relative_path(__file__, "test_cli_commands.py"),
             "module_name": None,
             "attribute": job_def_name,
         },
         {
             "workspace": None,
             "job_name": None,
-            "python_file": (file_relative_path(__file__, "test_cli_commands.py"),),
+            "python_file": file_relative_path(__file__, "test_cli_commands.py"),
             "module_name": None,
             "attribute": job_def_name,
             "working_directory": os.path.dirname(__file__),
         },
         {
             "workspace": None,
             "job_name": None,
-            "python_file": (file_relative_path(__file__, "test_cli_commands.py"),),
+            "python_file": file_relative_path(__file__, "test_cli_commands.py"),
             "module_name": None,
             "attribute": job_fn_name,
         },
     ]
 
 
+def job_python_args_to_workspace_args(args):
+    # Turn args expecting non-multiple files/modules into args allowing multiple
+    return [
+        {
+            "workspace": a.get("workspace"),
+            "job_name": a.get("job_name"),
+            "python_file": (a["python_file"],) if a.get("python_file") else None,
+            "module_name": (a["module_name"],) if a.get("module_name") else None,
+            "attribute": a["attribute"],
+            "package_name": a.get("package_name"),
+        }
+        for a in args
+    ]
+
+
 def valid_external_pipeline_target_args():
     return [
         {
@@ -593,7 +618,7 @@ def valid_external_pipeline_target_args():
             "module_name": None,
             "attribute": None,
         },
-    ] + [args for args in valid_job_python_origin_target_args()]
+    ] + job_python_args_to_workspace_args(valid_job_python_origin_target_args())
 
 
 def valid_pipeline_python_origin_target_cli_args():