[cli] Implement dagster definitions validate (#23692)

## Summary & Motivation

This PR implements `dagster definitions validate`, a command to validate
the definitions of a Dagster project.

---

### Expected behavior: 

Have a command to validate the definitions of a Dagster project. A
behavior similar to
[Definitions.validate_loadable](https://github.com/dagster-io/dagster/blob/e755c47bc8e1f596fdb4c4d68551732ee656dcfb/python_modules/dagster/dagster/_core/definitions/definitions_class.py#L632-L643)
is expected.

Example:

```
> dagster definitions validate -f path/to/my/definitions.py

Validating definitions in path/to/my/definitions.py
...

> dagster definitions validate -m my_dagster_project.definitions

Validating definitions in my_dagster_project.definitions.
...

> dagster definitions validate

Validating definitions in ...
...
```

### Input:

Similar to `dagster dev` as described
[here](https://docs.dagster.io/guides/running-dagster-locally#locating-your-code)
- the definitions can be validated:
- from a file, specified as a command line argument, eg.
`definitions.py`
- from a module, specified as a command line argument, eg.
`my_dagster_project.definitions`
- without command line arguments, read from `pyproject.toml` or
`workspace.yml`.

### Output:

Initially as logs in stdout/stderr.

### In scope:
- Support the validation of Dagster definitions defined in a
`Definitions` object.
- Implement a behavior similar to `Definitions.validate_loadable` where
the enclosed definitions will be loadable by Dagster:
  - No assets have conflicting keys.
  - No jobs, sensors, or schedules have conflicting names.
  - All asset jobs can be resolved.
  - All resource requirements are satisfied.

### Out of scope:
- Support the validation of Dagster definitions defined in a
`Repository` object.
- See comment
[here](https://dagsterlabs.slack.com/archives/C06C9QVJBUL/p1723762251730029?thread_ts=1723761841.463129&cid=C06C9QVJBUL)
- Validate definitions dependencies across code locations
- See thread
[here](https://dagsterlabs.slack.com/archives/C06C9QVJBUL/p1723642633214359)
  - Could be added in a next iteration.

### Potential improvement and next steps:
- Support usage in CI/CD
- Support output as JSON
- Support definitions constraints
- eg. Sensors' `minimum_interval_seconds` should always be of a at least
120 seconds.
- Support validation of definitions with dependencies across code
locations.


## How I Tested These Changes

Tested locally
BK with additional tests

## Changelog

Add the `dagster definitions validate` command to Dagster CLI. This
command validates if Dagster definitions are loadable.

- [x] `NEW` _(added new feature or capability)_
- [ ] `BUGFIX` _(fixed a bug)_
- [ ] `DOCS` _(added or updated documentation)_

Author: maximearmstrong

python_modules/dagster/dagster/_cli/__init__.py

@@ -4,6 +4,7 @@
 from dagster._cli.asset import asset_cli
 from dagster._cli.code_server import code_server_cli
 from dagster._cli.debug import debug_cli
+from dagster._cli.definitions import definitions_cli
 from dagster._cli.dev import dev_command
 from dagster._cli.instance import instance_cli
 from dagster._cli.job import job_cli
@@ -27,6 +28,7 @@ def create_dagster_cli():
         "project": project_cli,
         "dev": dev_command,
         "code-server": code_server_cli,
+        "definitions": definitions_cli,
     }
 
     @click.group(

python_modules/dagster/dagster/_cli/definitions.py

@@ -0,0 +1,78 @@
+import logging
+import sys
+
+import click
+
+from dagster import __version__ as dagster_version
+from dagster._cli.job import apply_click_params
+from dagster._cli.utils import get_possibly_temporary_instance_for_cli
+from dagster._cli.workspace.cli_target import (
+    ClickArgValue,
+    get_workspace_from_kwargs,
+    python_file_option,
+    python_module_option,
+    workspace_option,
+)
+from dagster._utils.log import configure_loggers
+
+
+@click.group(name="definitions")
+def definitions_cli():
+    """Commands for working with Dagster definitions."""
+
+
+def validate_command_options(f):
+    return apply_click_params(
+        f,
+        workspace_option(),
+        python_file_option(allow_multiple=True),
+        python_module_option(allow_multiple=True),
+    )
+
+
+@validate_command_options
+@click.option(
+    "--log-level",
+    help="Set the log level for dagster services.",
+    show_default=True,
+    default="info",
+    type=click.Choice(["critical", "error", "warning", "info", "debug"], case_sensitive=False),
+)
+@click.option(
+    "--log-format",
+    type=click.Choice(["colored", "json", "rich"], case_sensitive=False),
+    show_default=True,
+    required=False,
+    default="colored",
+    help="Format of the logs for dagster services",
+)
+@definitions_cli.command(
+    name="validate",
+    help="Validate if Dagster definitions are loadable.",
+)
+def definitions_validate_command(log_level: str, log_format: str, **kwargs: ClickArgValue):
+    configure_loggers(formatter=log_format, log_level=log_level.upper())
+    logger = logging.getLogger("dagster")
+
+    logger.info("Starting validation...")
+    with get_possibly_temporary_instance_for_cli(
+        "dagster definitions validate", logger=logger
+    ) as instance:
+        with get_workspace_from_kwargs(
+            instance=instance, version=dagster_version, kwargs=kwargs
+        ) as workspace:
+            invalid = any(
+                entry
+                for entry in workspace.get_code_location_entries().values()
+                if entry.load_error
+            )
+            for code_location, entry in workspace.get_code_location_entries().items():
+                if entry.load_error:
+                    logger.error(
+                        f"Validation failed for code location {code_location} with exception: "
+                        f"{entry.load_error.message}."
+                    )
+                else:
+                    logger.info(f"Validation successful for code location {code_location}.")
+    logger.info("Ending validation...")
+    sys.exit(0) if not invalid else sys.exit(1)

python_modules/dagster/dagster_tests/cli_tests/command_tests/definitions_command_projects/empty_project/empty_project/__init__.py

@@ -0,0 +1,14 @@
+from dagster import Definitions, asset, define_asset_job
+
+
+@asset(name="my_asset")
+def my_first_asset() -> None: ...
+
+
+@asset(name="my_asset")
+def my_second_asset() -> None: ...
+
+
+my_job = define_asset_job(name="my_job", selection="my_asset")
+
+defs = Definitions(assets=[my_first_asset, my_second_asset], jobs=[my_job])

python_modules/dagster/dagster_tests/cli_tests/command_tests/definitions_command_projects/invalid_project/invalid_project/__init__.py

@@ -0,0 +1,2 @@
+[tool.dagster]
+module_name = "invalid_project.definitions"

python_modules/dagster/dagster_tests/cli_tests/command_tests/definitions_command_projects/invalid_project/invalid_project/definitions.py

@@ -0,0 +1,3 @@
+load_from:
+  - python_file:
+      relative_path: invalid_project/definitions.py

python_modules/dagster/dagster_tests/cli_tests/command_tests/definitions_command_projects/invalid_project/pyproject.toml

@@ -0,0 +1,2 @@
+[tool.dagster]
+module_name = "valid_project.definitions"

python_modules/dagster/dagster_tests/cli_tests/command_tests/definitions_command_projects/invalid_project/workspace.yaml

@@ -0,0 +1 @@
+

python_modules/dagster/dagster_tests/cli_tests/command_tests/definitions_command_projects/valid_project/pyproject.toml

@@ -0,0 +1,10 @@
+from dagster import Definitions, asset, define_asset_job
+
+
+@asset
+def my_asset() -> None: ...
+
+
+my_job = define_asset_job(name="my_job", selection="my_asset")
+
+defs = Definitions(assets=[my_asset], jobs=[my_job])

python_modules/dagster/dagster_tests/cli_tests/command_tests/definitions_command_projects/valid_project/valid_project/__init__.py

@@ -0,0 +1,10 @@
+from dagster import Definitions, asset, define_asset_job
+
+
+@asset
+def my_other_asset() -> None: ...
+
+
+my_other_job = define_asset_job(name="my_other_job", selection="my_other_asset")
+
+defs = Definitions(assets=[my_other_asset], jobs=[my_other_job])

python_modules/dagster/dagster_tests/cli_tests/command_tests/definitions_command_projects/valid_project/valid_project/definitions.py

@@ -0,0 +1,3 @@
+load_from:
+  - python_file:
+      relative_path: valid_project/definitions.py

python_modules/dagster/dagster_tests/cli_tests/command_tests/definitions_command_projects/valid_project/valid_project/more_definitions.py

@@ -0,0 +1,72 @@
+from typing import Optional, Sequence
+
+import pytest
+from click.testing import CliRunner
+from dagster._cli.definitions import definitions_validate_command
+from dagster._utils import file_relative_path
+
+EMPTY_PROJECT_PATH = file_relative_path(__file__, "definitions_command_projects/empty_project")
+VALID_PROJECT_PATH = file_relative_path(__file__, "definitions_command_projects/valid_project")
+INVALID_PROJECT_PATH = file_relative_path(__file__, "definitions_command_projects/invalid_project")
+
+
+def invoke_validate(options: Optional[Sequence[str]] = None):
+    runner = CliRunner()
+    return runner.invoke(definitions_validate_command, options)
+
+
+def test_empty_project(monkeypatch):
+    with monkeypatch.context() as m:
+        m.chdir(EMPTY_PROJECT_PATH)
+        result = invoke_validate()
+        assert result.exit_code == 2
+        assert (
+            "Error: No arguments given and no [tool.dagster] block in pyproject.toml found."
+            in result.output
+        )
+
+
+@pytest.mark.parametrize(
+    "options",
+    [
+        [],
+        ["-f", "valid_project/definitions.py"],
+        ["-f", "valid_project/definitions.py"],
+        ["-m", "valid_project.definitions"],
+        ["-w", "workspace.yaml"],
+    ],
+)
+def test_valid_project(options, monkeypatch):
+    with monkeypatch.context() as m:
+        m.chdir(VALID_PROJECT_PATH)
+        result = invoke_validate(options=options)
+        assert result.exit_code == 0
+        assert "Validation successful" in result.output
+
+
+def test_valid_project_with_multiple_definitions_files(monkeypatch):
+    with monkeypatch.context() as m:
+        m.chdir(VALID_PROJECT_PATH)
+        options = ["-f", "valid_project/definitions.py", "-f", "valid_project/more_definitions.py"]
+        result = invoke_validate(options=options)
+        assert result.exit_code == 0
+        assert "Validation successful for code location definitions.py." in result.output
+        assert "Validation successful for code location more_definitions.py." in result.output
+
+
+@pytest.mark.parametrize(
+    "options",
+    [
+        [],
+        ["-f", "invalid_project/definitions.py"],
+        ["-m", "invalid_project.definitions"],
+        ["-w", "workspace.yaml"],
+    ],
+)
+def test_invalid_project(options, monkeypatch):
+    with monkeypatch.context() as m:
+        m.chdir(INVALID_PROJECT_PATH)
+        result = invoke_validate(options=options)
+        assert result.exit_code == 1
+        assert "Validation failed" in result.output
+        assert "Duplicate asset key: AssetKey(['my_asset'])" in result.output