dg deploy initial serverless + docker implementation (#28704)

Lots to do here as evidenced by the many TODOs, but this is sufficient to deploy a dg project to serverless. It:

- checks that you are logged in
- Creates a dagster_cloud.yaml (only used under the hood) derived from the pyproject.toml for the project
- scaffolds a uv-friendly Dockerfile if none exists
- Builds the project using that Dockerfile and pushes it to the remote serverless repository
- Deploys the project, at which point it will be available in Dagster+.

Also would like to improve the testing coverage here with a real test harness that deploys to a locally available cloud instance (just mocking out the call to the dagster-cloud ci leaves a lot of things untested, although each of those underlying commands also have separate unite tests)

Author: gibsondan

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

@@ -1,12 +1,19 @@
+import os
+import sys
+import tempfile
 import webbrowser
 from collections.abc import Mapping
+from contextlib import ExitStack
 from pathlib import Path
+from typing import Optional
 
 import click
+import jinja2
 from dagster_shared.plus.config import DagsterPlusCliConfig
 from dagster_shared.plus.login_server import start_login_server
 
 from dagster_dg.cli.shared_options import dg_global_options
+from dagster_dg.cli.utils import create_temp_dagster_cloud_yaml_file
 from dagster_dg.config import normalize_cli_config
 from dagster_dg.context import DgContext
 from dagster_dg.env import ProjectEnvVars
@@ -144,3 +151,130 @@ def pull_env_command(**global_options: object) -> None:
             click.echo(
                 f"Environment variables not found for projects: {', '.join(projects_without_secrets)}"
             )
+
+
+def _create_temp_deploy_dockerfile(dst_path, python_version):
+    dockerfile_template_path = (
+        Path(__file__).parent.parent / "templates" / "deploy_uv_Dockerfile.jinja"
+    )
+
+    loader = jinja2.FileSystemLoader(searchpath=os.path.dirname(dockerfile_template_path))
+    env = jinja2.Environment(loader=loader)
+
+    template = env.get_template(os.path.basename(dockerfile_template_path))
+
+    with open(dst_path, "w", encoding="utf8") as f:
+        f.write(template.render(python_version=python_version))
+        f.write("\n")
+
+
+@plus_group.command(name="deploy", cls=DgClickCommand)
+@click.option(
+    "--organization",
+    "organization",
+    help="Dagster+ organization to which to deploy. If not set, defaults to the value set by `dg plus login`.",
+    envvar="DAGSTER_PLUS_ORGANIZATION",
+)
+@click.option(
+    "--deployment",
+    "deployment",
+    help="Name of the Dagster+ deployment to which to deploy. If not set, defaults to the value set by `dg plus login`.",
+    envvar="DAGSTER_PLUS_DEPLOYMENT",
+)
+@click.option(
+    "--python-version",
+    "python_version",
+    type=click.Choice(["3.9", "3.10", "3.11", "3.12"]),
+    help=(
+        "Python version used to deploy the project. If not set, defaults to the calling process's Python minor version."
+    ),
+)
+@dg_global_options
+@cli_telemetry_wrapper
+def deploy_command(
+    organization: Optional[str],
+    deployment: Optional[str],
+    python_version: Optional[str],
+    **global_options: object,
+) -> None:
+    """Deploy a project to Dagster Plus."""
+    cli_config = normalize_cli_config(global_options, click.get_current_context())
+
+    if not python_version:
+        python_version = f"3.{sys.version_info.minor}"
+
+    plus_config = DagsterPlusCliConfig.get()
+
+    organization = organization or plus_config.organization
+    if not organization:
+        raise click.UsageError(
+            "Organization not specified. To specify an organization, use the --organization option "
+            "or run `dg plus login`."
+        )
+
+    deployment = deployment or plus_config.default_deployment
+    if not deployment:
+        raise click.UsageError(
+            "Deployment not specified. To specify a deployment, use the --deployment option "
+            "or run `dg plus login`."
+        )
+
+    # TODO This command should work in a workspace context too and apply to multiple projects
+    dg_context = DgContext.for_project_environment(Path.cwd(), cli_config)
+
+    # TODO Confirm that dagster-cloud is packaged in the project
+
+    with ExitStack() as stack:
+        # TODO Once this is split out into multiple commands, we need a default statedir
+        # that can be persisted across commands.
+        statedir = stack.enter_context(tempfile.TemporaryDirectory())
+
+        # Construct a dagster_cloud.yaml file based on info in the pyproject.toml
+        dagster_cloud_yaml_file = stack.enter_context(
+            create_temp_dagster_cloud_yaml_file(dg_context)
+        )
+
+        dg_context.external_dagster_cloud_cli_command(
+            [
+                "ci",
+                "init",
+                "--statedir",
+                str(statedir),
+                "--dagster-cloud-yaml-path",
+                dagster_cloud_yaml_file,
+                "--project-dir",
+                str(dg_context.root_path),
+                "--deployment",
+                deployment,
+                "--organization",
+                organization,
+            ],
+        )
+
+        dockerfile_path = dg_context.root_path / "Dockerfile"
+        if not os.path.exists(dockerfile_path):
+            click.echo(f"No Dockerfile found - scaffolding a default one at {dockerfile_path}.")
+            _create_temp_deploy_dockerfile(dockerfile_path, python_version)
+        else:
+            click.echo(f"Building using Dockerfile at {dockerfile_path}.")
+
+        # TODO This command is serverless-specific, support hybrid as well
+        dg_context.external_dagster_cloud_cli_command(
+            [
+                "ci",
+                "build",
+                "--statedir",
+                str(statedir),
+                "--dockerfile-path",
+                str(dg_context.root_path / "Dockerfile"),
+            ],
+        )
+
+        dg_context.external_dagster_cloud_cli_command(
+            [
+                "ci",
+                "deploy",
+                "--statedir",
+                str(statedir),
+            ],
+        )

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

@@ -176,6 +176,32 @@ def create_temp_workspace_file(dg_context: DgContext) -> Iterator[str]:
         yield temp_workspace_file.name
 
 
+def _dagster_cloud_entry_for_project(dg_context: DgContext) -> dict[str, Any]:
+    return {
+        "location_name": dg_context.code_location_name,
+        "code_source": {
+            "module_name": str(dg_context.code_location_target_module_name),
+        },
+        # TODO build, container_context
+    }
+
+
+@contextmanager
+def create_temp_dagster_cloud_yaml_file(dg_context: DgContext) -> Iterator[str]:
+    with NamedTemporaryFile(mode="w+", delete=True) as temp_dagster_cloud_yaml_file:
+        entries = []
+        if dg_context.is_project:
+            entries.append(_dagster_cloud_entry_for_project(dg_context))
+        elif dg_context.is_workspace:
+            for spec in dg_context.project_specs:
+                project_root = dg_context.root_path / spec.path
+                project_context: DgContext = dg_context.with_root_path(project_root)
+                entries.append(_dagster_cloud_entry_for_project(project_context))
+        yaml.dump({"locations": entries}, temp_dagster_cloud_yaml_file)
+        temp_dagster_cloud_yaml_file.flush()
+        yield temp_dagster_cloud_yaml_file.name
+
+
 class DagsterCliCmd(NamedTuple):
     cmd_location: str
     cmd: list[str]

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

@@ -434,6 +434,20 @@ def _dagster_components_entry_points(self) -> Mapping[str, str]:
     # ##### HELPERS
     # ########################
 
+    def external_dagster_cloud_cli_command(
+        self, command: list[str], log: bool = True, env: Optional[dict[str, str]] = None
+    ):
+        # TODO Match dagster-cloud-cli version with calling dg version
+        command = ["uv", "tool", "run", "--from", "dagster-cloud-cli", "dagster-cloud", *command]
+        with pushd(self.root_path):
+            result = subprocess.run(command, check=False, env={**os.environ, **(env or {})})
+            if result.returncode != 0:
+                exit_with_error(f"""
+                    An error occurred while executing a `dagster-cloud` command via `uv tool run`.
+
+                    `{shlex.join(command)}` exited with code {result.returncode}. Aborting.
+                """)
+
     def external_components_command(
         self,
         command: list[str],

python_modules/libraries/dagster-dg/dagster_dg/templates/deploy_uv_Dockerfile.jinja

@@ -0,0 +1,33 @@
+# Based on: https://github.com/astral-sh/uv-docker-example/blob/main/multistage.Dockerfile
+
+# First, build the application in the `/app` directory.
+FROM ghcr.io/astral-sh/uv:python{{ python_version }}-bookworm-slim AS builder
+ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy
+
+# Disable Python downloads, because we want to use the system interpreter
+# across both images. If using a managed Python version, it needs to be
+# copied from the build image into the final image; see `standalone.Dockerfile`
+# for an example.
+ENV UV_PYTHON_DOWNLOADS=0
+
+WORKDIR /app
+RUN --mount=type=cache,target=/root/.cache/uv \
+    --mount=type=bind,source=uv.lock,target=uv.lock \
+    --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
+    uv sync --frozen --no-install-project --no-dev
+ADD . /app
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --frozen --no-dev
+
+
+# Then, use a final image without uv
+FROM python:{{ python_version }}-slim-bookworm
+# It is important to use the image that matches the builder, as the path to the
+# Python executable must be the same, e.g., using `python:3.11-slim-bookworm`
+# will fail.
+
+# Copy the application from the builder
+COPY --from=builder --chown=app:app /app /app
+
+# Place executables in the environment at the front of the path
+ENV PATH="/app/.venv/bin:$PATH"

python_modules/libraries/dagster-dg/dagster_dg_tests/cli_tests/plus_tests/__init__.py

@@ -0,0 +1,70 @@
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from click.testing import CliRunner
+from dagster_dg.cli.plus import plus_group
+from dagster_shared.plus.config import DagsterPlusCliConfig
+
+from dagster_dg_tests.utils import isolated_example_project_foo_bar
+
+
+@pytest.fixture
+def logged_in_dg_cli_config(empty_dg_cli_config):
+    config = DagsterPlusCliConfig(
+        organization="hooli",
+        user_token="fake-user-token",
+        default_deployment="prod",
+    )
+    config.write()
+    yield
+
+
+@pytest.fixture
+def empty_dg_cli_config(monkeypatch):
+    with (
+        tempfile.TemporaryDirectory() as tmp_dg_dir,
+    ):
+        config_path = Path(tmp_dg_dir) / "dg.toml"
+        monkeypatch.setenv("DG_CLI_CONFIG", config_path)
+        config = DagsterPlusCliConfig(
+            organization="",
+            user_token="",
+            default_deployment="",
+        )
+        config.write()
+        yield config_path
+
+
+@pytest.fixture(scope="module")
+def runner():
+    yield CliRunner()
+
+
+@pytest.fixture(scope="module")
+def project(runner):
+    with isolated_example_project_foo_bar(runner, use_editable_dagster=False, in_workspace=False):
+        yield
+
+
+def test_plus_deploy_command(logged_in_dg_cli_config, project, runner):
+    with patch(
+        "dagster_dg.context.DgContext.external_dagster_cloud_cli_command",
+    ):
+        result = runner.invoke(plus_group, ["deploy"])
+        assert result.exit_code == 0, result.output + " : " + str(result.exception)
+        assert "No Dockerfile found - scaffolding a default one" in result.output
+
+        result = runner.invoke(plus_group, ["deploy"])
+        assert "Building using Dockerfile at" in result.output
+        assert result.exit_code == 0, result.output + " : " + str(result.exception)
+
+
+def test_plus_deploy_command_no_login(empty_dg_cli_config, runner, project):
+    with patch(
+        "dagster_dg.context.DgContext.external_dagster_cloud_cli_command",
+    ):
+        result = runner.invoke(plus_group, ["deploy"])
+        assert result.exit_code != 0
+        assert "Organization not specified" in result.output

python_modules/libraries/dagster-dg/dagster_dg_tests/cli_tests/plus_tests/test_plus_deploy_command.py

@@ -17,9 +17,10 @@
 from dagster_dg.utils.plus import gql
 
 ensure_dagster_dg_tests_import()
-from dagster_dg_tests.cli_tests.plus_tests.utils import mock_gql_response
 from dagster_shared.plus.config import DagsterPlusCliConfig
 
+from dagster_dg_tests.cli_tests.plus_tests.utils import mock_gql_response
+
 
 @pytest.fixture()
 def setup_cloud_cli_config(monkeypatch):

python_modules/libraries/dagster-dg/dagster_dg_tests/cli_tests/plus_tests/test_plus_login_command.py

@@ -198,6 +198,7 @@ def isolated_example_project_foo_bar(
     component_dirs: Sequence[Path] = [],
     config_file_type: ConfigFileType = "pyproject.toml",
     package_layout: PackageLayoutType = "src",
+    use_editable_dagster: bool = True,
 ) -> Iterator[None]:
     """Scaffold a project named foo_bar in an isolated filesystem.
 
@@ -218,8 +219,7 @@ def isolated_example_project_foo_bar(
         args = [
             "scaffold",
             "project",
-            "--use-editable-dagster",
-            dagster_git_repo_dir,
+            *(["--use-editable-dagster", dagster_git_repo_dir] if use_editable_dagster else []),
             *(["--skip-venv"] if skip_venv else []),
             *(["--no-populate-cache"] if not populate_cache else []),
             "foo-bar",