misc fixes

Author: benpankow

python_modules/libraries/dagster-components/dagster_components/cli/list.py

@@ -70,6 +70,7 @@ def list_all_components_schema_command(entry_points: bool, extra_modules: tuple[
                     key.name,
                     type=(Literal[key_string], key_string),
                     attributes=(schema_type, None),
+                    requires=(dict[str, list[str]], None),
                     __config__=ConfigDict(extra="forbid"),
                 )
             )

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

@@ -196,8 +196,7 @@ def check_yaml(
             )
         if updated_files:
             click.echo(
-                "The following component files were updated to fix environment variable requirements:\n"
-                + "\n".join(updated_files)
+                f"{len(updated_files)} component files were updated to fix environment variable requirements."
             )
         return False
     else:
@@ -206,10 +205,8 @@ def check_yaml(
         ) - os.environ.keys()
         if updated_files:
             click.echo(
-                "The following component files were updated to fix environment variable requirements:\n"
-                + "\n".join(str(file) for file in updated_files)
+                f"{len(updated_files)} component files were updated to fix environment variable requirements."
             )
-
         if missing_env_vars:
             click.echo(
                 "The following environment variables are used in components but not specified in the .env file or the current shell environment:\n"

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

@@ -4,6 +4,7 @@
 
 from dagster_dg.cache import DgCache
 from dagster_dg.cli.check import check_group
+from dagster_dg.cli.deploy import deploy_command
 from dagster_dg.cli.dev import dev_command
 from dagster_dg.cli.docs import docs_group
 from dagster_dg.cli.env import env_group
@@ -37,6 +38,7 @@ def create_dg_cli():
             "dev": dev_command,
             "init": init_command,
             "plus": plus_group,
+            "deploy": deploy_command,
         },
         context_settings={
             "max_content_width": DG_CLI_MAX_OUTPUT_WIDTH,

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

@@ -0,0 +1,68 @@
+from pathlib import Path
+
+import click
+from dagster_shared.plus.config import DagsterPlusCliConfig
+
+from dagster_dg.cli.shared_options import dg_global_options
+from dagster_dg.config import normalize_cli_config
+from dagster_dg.context import DgContext
+from dagster_dg.env import ProjectEnvVars, get_project_specified_env_vars
+from dagster_dg.utils import DgClickCommand
+from dagster_dg.utils.plus import gql
+from dagster_dg.utils.plus.gql_client import DagsterCloudGraphQLClient
+from dagster_dg.utils.telemetry import cli_telemetry_wrapper
+
+# ########################
+# ##### ENVIRONMENT
+# ########################
+
+
+@click.command(name="deploy", cls=DgClickCommand)
+@dg_global_options
+@cli_telemetry_wrapper
+def deploy_command(**global_options: object) -> None:
+    """Deploy a Dagster project."""
+    cli_config = normalize_cli_config(global_options, click.get_current_context())
+    dg_context = DgContext.for_project_environment(Path.cwd(), cli_config)
+
+    env_vars = get_project_specified_env_vars(dg_context)
+
+    env = ProjectEnvVars.from_ctx(dg_context)
+
+    env_var_keys = set(env.values.keys()) | set(env_vars.keys())
+
+    if DagsterPlusCliConfig.exists():
+        scopes_for_key = {}
+        config = DagsterPlusCliConfig.get()
+        gql_client = DagsterCloudGraphQLClient.from_config(config)
+        for key in env_var_keys:
+            secrets_by_location = gql_client.execute(
+                gql.GET_SECRETS_FOR_SCOPES_QUERY,
+                {
+                    "locationName": dg_context.project_name,
+                    "scopes": {
+                        "fullDeploymentScope": True,
+                        "allBranchDeploymentsScope": True,
+                        "localDeploymentScope": True,
+                    },
+                    "secretName": key,
+                },
+            )["secretsForScopes"]["secrets"]
+            scopes_for_key[key] = {
+                "full": any(secret["fullDeploymentScope"] for secret in secrets_by_location),
+                "branch": any(
+                    secret["allBranchDeploymentsScope"] for secret in secrets_by_location
+                ),
+                "local": any(secret["localDeploymentScope"] for secret in secrets_by_location),
+            }
+
+    if not all(scopes_for_key[key]["full"] for key in env_var_keys):
+        click.echo(
+            f"Missing environment variables for deployment {config.default_deployment} in full deployment scope, for location {dg_context.project_name}:"
+        )
+        for key in env_var_keys:
+            if not scopes_for_key[key]["full"]:
+                click.echo(f"  {key}")
+        import sys
+
+        sys.exit(1)

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

@@ -146,6 +146,7 @@ def dev_command(
                 check_result = check_yaml_fn(
                     dg_context.for_project_environment(project_dir, cli_config),
                     [],
+                    fix_env_requirements=False,
                 )
                 overall_check_result = overall_check_result and check_result
             if not overall_check_result:

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

@@ -3,6 +3,7 @@
 
 import click
 import yaml
+from dagster_shared.plus.config import DagsterPlusCliConfig
 from dagster_shared.yaml_utils import parse_yaml_with_source_positions
 from rich.console import Console
 from rich.table import Table
@@ -12,8 +13,10 @@
 from dagster_dg.component import get_specified_env_var_deps, get_used_env_vars
 from dagster_dg.config import normalize_cli_config
 from dagster_dg.context import DgContext
-from dagster_dg.env import ProjectEnvVars
+from dagster_dg.env import ProjectEnvVars, get_project_specified_env_vars
 from dagster_dg.utils import DgClickCommand, DgClickGroup
+from dagster_dg.utils.plus import gql
+from dagster_dg.utils.plus.gql_client import DagsterCloudGraphQLClient
 from dagster_dg.utils.telemetry import cli_telemetry_wrapper
 
 
@@ -35,16 +38,61 @@ def list_env_command(**global_options: object) -> None:
     cli_config = normalize_cli_config(global_options, click.get_current_context())
     dg_context = DgContext.for_project_environment(Path.cwd(), cli_config)
 
+    env_vars = get_project_specified_env_vars(dg_context)
+
     env = ProjectEnvVars.from_ctx(dg_context)
-    if not env.values:
-        click.echo("No environment variables are defined for this project.")
-        return
 
     table = Table(border_style="dim")
     table.add_column("Env Var")
-    table.add_column("Value")
-    for key, value in env.values.items():
-        table.add_row(key, value)
+    table.add_column("Local Value")
+    table.add_column("Components")
+    env_var_keys = set(env.values.keys()) | set(env_vars.keys())
+
+    if DagsterPlusCliConfig.exists():
+        table.add_column("Dev")
+        table.add_column("Branch")
+        table.add_column("Full")
+        scopes_for_key = {}
+        config = DagsterPlusCliConfig.get()
+        gql_client = DagsterCloudGraphQLClient.from_config(config)
+        for key in env_var_keys:
+            secrets_by_location = gql_client.execute(
+                gql.GET_SECRETS_FOR_SCOPES_QUERY,
+                {
+                    "locationName": dg_context.project_name,
+                    "scopes": {
+                        "fullDeploymentScope": True,
+                        "allBranchDeploymentsScope": True,
+                        "localDeploymentScope": True,
+                    },
+                    "secretName": key,
+                },
+            )["secretsForScopes"]["secrets"]
+            scopes_for_key[key] = {
+                "full": any(secret["fullDeploymentScope"] for secret in secrets_by_location),
+                "branch": any(
+                    secret["allBranchDeploymentsScope"] for secret in secrets_by_location
+                ),
+                "local": any(secret["localDeploymentScope"] for secret in secrets_by_location),
+            }
+
+    if not env_var_keys:
+        click.echo("No environment variables are defined for this project.")
+        return
+
+    for key in env_var_keys:
+        value = env.values.get(key)
+        components = env_vars.get(key, [])
+        table.add_row(
+            key,
+            value,
+            ", ".join(str(path) for path in components),
+            *(
+                ["X" if scopes_for_key[key][scope] else "" for scope in ["local", "branch", "full"]]
+                if DagsterPlusCliConfig.exists()
+                else []
+            ),
+        )
     console = Console()
     console.print(table)
 

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

@@ -141,7 +141,8 @@ def pull_env_command(**global_options: object) -> None:
                 secrets_by_location[project_ctx.project_name]
             )
             env.write()
-            projects_without_secrets.remove(project_ctx.project_name)
+            if project_ctx.project_name in projects_without_secrets:
+                projects_without_secrets.remove(project_ctx.project_name)
 
     if dg_context.is_project:
         click.echo("Environment variables saved to .env")

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

@@ -1,13 +1,38 @@
-from collections.abc import Mapping
+from collections import defaultdict
+from collections.abc import Mapping, Sequence
+from pathlib import Path
 from typing import Optional
 
 from dagster_shared import check
+from dagster_shared.yaml_utils import parse_yaml_with_source_positions
 from dotenv import dotenv_values
 from typing_extensions import Self
+from yaml.scanner import ScannerError
 
+from dagster_dg.component import get_specified_env_var_deps
 from dagster_dg.context import DgContext
 
 
+def get_project_specified_env_vars(dg_context: DgContext) -> Mapping[str, Sequence[Path]]:
+    env_vars = defaultdict(list)
+    for component_dir in dg_context.defs_path.iterdir():
+        component_path = component_dir / "component.yaml"
+
+        if component_path.exists():
+            text = component_path.read_text()
+            try:
+                component_doc_tree = parse_yaml_with_source_positions(
+                    text, filename=str(component_path)
+                )
+            except ScannerError:
+                continue
+
+            specified_env_var_deps = get_specified_env_var_deps(component_doc_tree.value)
+            for key in specified_env_var_deps:
+                env_vars[key].append(component_path.relative_to(dg_context.defs_path).parent)
+    return env_vars
+
+
 class ProjectEnvVars:
     """Represents the environment for a project, stored in the .env file of a
     project root.

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

@@ -59,7 +59,7 @@ def install_or_update_yaml_schema_extension(
     template_package_json_path = Path(__file__).parent / "vscode_extension_package.json"
     template_package_json = json.loads(template_package_json_path.read_text())
     template_package_json["contributes"]["yamlValidation"] = [
-        {"fileMatch": f"{yaml_dir}/**/*.y*ml", "url": f"{schema_path}"}
+        {"fileMatch": f"{yaml_dir}/**/component.y*ml", "url": f"{schema_path}"}
     ]
 
     extension_working_dir.mkdir(parents=True, exist_ok=True)