[components] Add windows testing to dagster-dg (BUILD-672)

azure-pipelines.yml

@@ -9,7 +9,7 @@ parameters:
     type: object
     default:
       - "3.10"
-  - name: py3_env_suffixes
+  - name: dagster_core_py3_env_suffixes
     type: object
     default:
       - api_tests
@@ -26,23 +26,37 @@ jobs:
     strategy:
       matrix:
         ${{ each py_version in parameters.py3_versions }}:
-          ${{ each env_suffix in parameters.py3_env_suffixes }}:
+          ${{ each env_suffix in parameters.dagster_core_py3_env_suffixes }}:
             ${{ replace(py_version, '.', '') }}-windows-${{ env_suffix }}:
-              TOXENV: "py${{ replace(py_version, '.', '') }}-windows-${{ env_suffix }}"
-              python.version: "${{ py_version }}"
+              PACKAGE_ROOT: "python_modules\\dagster"
+              PATH_BACK_TO_REPO_ROOT: "..\\.."
+              TOX_ENV: "py${{ replace(py_version, '.', '') }}-windows-${{ env_suffix }}"
+              PYTHON_VERSION: "${{ py_version }}"
+
+        ${{ each py_version in parameters.py3_versions }}:
+          dagster-dg-${{ replace(py_version, '.', '') }}:
+            PACKAGE_ROOT: "python_modules\\libraries\\dagster-dg"
+            PATH_BACK_TO_REPO_ROOT: "..\\..\\.."
+            TOX_ENV: windows
+            PYTHON_VERSION: "${{ py_version }}"
     variables:
-      PYTHONLEGACYWINDOWSSTDIO: "1"
+      PYTHONUTF8: "1"
+    # Use PowerShell (`powershell`) instead of cmd shell (`script`) for better UTF-8 support.
     steps:
       - task: UsePythonVersion@0
         inputs:
-          versionSpec: "$(python.version)"
+          versionSpec: "$(PYTHON_VERSION)"
           architecture: "x64"
-      - script: pip install "tox<4.0.0" uv
+      - powershell: |
+          pip install "tox<4.0.0" uv
         displayName: "Install tox & uv"
-      - script: cd python_modules\dagster && tox -e %TOXENV% && cd ..\..
+      - powershell: |
+          cd $env:PACKAGE_ROOT
+          tox -e $env:TOX_ENV
+          cd $env:PATH_BACK_TO_REPO_ROOT
         displayName: "Run tests"
       - task: PublishTestResults@2
         inputs:
           testResultsFiles: "**/test_results.xml"
-          testRunTitle: "dagster $(TOXENV)"
+          testRunTitle: "$(PACKAGE_ROOT) $(TOX_ENV)"
         condition: succeededOrFailed()

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

@@ -1,9 +1,8 @@
 import hashlib
 import shutil
-import sys
 from collections.abc import Sequence
 from pathlib import Path
-from typing import Final, Literal, Optional, Union
+from typing import Final, Literal, Optional
 
 from typing_extensions import Self, TypeAlias
 
@@ -12,19 +11,19 @@
     DEFAULT_FILE_EXCLUDE_PATTERNS,
     hash_directory_metadata,
     hash_file_metadata,
+    is_macos,
+    is_windows,
 )
 
 _CACHE_CONTAINER_DIR_NAME: Final = "dg-cache"
 
-CachableDataType: TypeAlias = Union[
-    Literal["component_registry_data"], Literal["all_components_schema"]
-]
+CachableDataType: TypeAlias = Literal["component_registry_data", "all_components_schema"]
 
 
 def get_default_cache_dir() -> Path:
-    if sys.platform == "win32":
+    if is_windows():
         return Path.home() / "AppData" / "dg" / "cache"
-    elif sys.platform == "darwin":
+    elif is_macos():
         return Path.home() / "Library" / "Caches" / "dg"
     else:
         return Path.home() / ".cache" / "dg"
@@ -78,6 +77,7 @@ def clear_key(self, key: tuple[str, ...]) -> None:
 
     def clear_all(self) -> None:
         shutil.rmtree(self._root_path)
+        assert not self._root_path.exists()
         self.log(f"CACHE [clear-all]: {self._root_path}")
 
     def get(self, key: tuple[str, ...]) -> Optional[str]:

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

@@ -1,9 +1,6 @@
-import os
-import signal
 import subprocess
-import sys
 import time
-from collections.abc import Iterator, Mapping, Sequence
+from collections.abc import Iterator, Mapping
 from contextlib import contextmanager, nullcontext
 from pathlib import Path
 from tempfile import NamedTemporaryFile
@@ -17,7 +14,13 @@
 from dagster_dg.config import normalize_cli_config
 from dagster_dg.context import DgContext
 from dagster_dg.error import DgError
-from dagster_dg.utils import DgClickCommand, exit_with_error, pushd
+from dagster_dg.utils import (
+    DgClickCommand,
+    exit_with_error,
+    interrupt_subprocess,
+    open_subprocess,
+    pushd,
+)
 
 T = TypeVar("T")
 
@@ -132,7 +135,7 @@ def dev_command(
     with pushd(dg_context.root_path), temp_workspace_file_cm as workspace_file:
         if workspace_file:  # only non-None deployment context
             cmd.extend(["--workspace", workspace_file])
-        uv_run_dagster_dev_process = _open_subprocess(cmd)
+        uv_run_dagster_dev_process = open_subprocess(cmd)
         try:
             while True:
                 time.sleep(_CHECK_SUBPROCESS_INTERVAL)
@@ -151,7 +154,7 @@ def dev_command(
             # directly to the `dagster dev` process (the only child of the `uv run` process). This
             # will cause `dagster dev` to terminate which in turn will cause `uv run` to terminate.
             dagster_dev_pid = _get_child_process_pid(uv_run_dagster_dev_process)
-            _interrupt_subprocess(dagster_dev_pid)
+            interrupt_subprocess(dagster_dev_pid)
 
             try:
                 uv_run_dagster_dev_process.wait(timeout=10)
@@ -162,7 +165,11 @@ def dev_command(
 
 @contextmanager
 def _temp_workspace_file(dg_context: DgContext) -> Iterator[str]:
-    with NamedTemporaryFile(mode="w+", delete=True) as temp_workspace_file:
+    # Note that we can't rely on delete=True here because the NamedTemporaryFile context manager
+    # will create a file lock on Windows that will prevent the child `dagster dev` process we spawn
+    # from being able to read the file. So we use delete=False, exit the context manager before
+    # yielding,  and manually delete the file after the context manager exits.
+    with NamedTemporaryFile(mode="w+", delete=False) as temp_workspace_file:
         entries = []
         for location in dg_context.get_code_location_names():
             code_location_root = dg_context.get_code_location_path(location)
@@ -177,39 +184,19 @@ def _temp_workspace_file(dg_context: DgContext) -> Iterator[str]:
             entries.append({"python_file": entry})
         yaml.dump({"load_from": entries}, temp_workspace_file)
         temp_workspace_file.flush()
+    try:
         yield temp_workspace_file.name
+    finally:
+        Path(temp_workspace_file.name).unlink()
 
 
 def _format_forwarded_option(option: str, value: object) -> list[str]:
     return [] if value is None else [option, str(value)]
 
 
-def _get_child_process_pid(proc: "subprocess.Popen") -> int:
-    children = psutil.Process(proc.pid).children(recursive=False)
+def _get_child_process_pid(proc: subprocess.Popen) -> int:
+    # Windows will sometimes return the parent process as its own child. Filter this out.
+    children = [p for p in psutil.Process(proc.pid).children(recursive=False) if p.pid != proc.pid]
     if len(children) != 1:
         raise ValueError(f"Expected exactly one child process, but found {len(children)}")
     return children[0].pid
-
-
-# Windows subprocess termination utilities. See here for why we send CTRL_BREAK_EVENT on Windows:
-# https://stefan.sofa-rockers.org/2013/08/15/handling-sub-process-hierarchies-python-linux-os-x/
-
-
-def _interrupt_subprocess(pid: int) -> None:
-    """Send CTRL_BREAK_EVENT on Windows, SIGINT on other platforms."""
-    if sys.platform == "win32":
-        os.kill(pid, signal.CTRL_BREAK_EVENT)
-    else:
-        os.kill(pid, signal.SIGINT)
-
-
-def _open_subprocess(command: Sequence[str]) -> "subprocess.Popen":
-    """Sets the correct flags to support graceful termination."""
-    creationflags = 0
-    if sys.platform == "win32":
-        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
-
-    return subprocess.Popen(
-        command,
-        creationflags=creationflags,
-    )

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

@@ -25,7 +25,7 @@ def install_completion(context: click.Context):
     shell, _ = shellingham.detect_shell()
     comp_class = click.shell_completion.get_completion_class(shell)
     if comp_class is None:
-        exit_with_error(f"Shell {shell} is not supported.")
+        exit_with_error(f"Shell `{shell}` is not supported.")
     else:
         comp_inst = comp_class(
             cli=context.command, ctx_args={}, prog_name="dg", complete_var="_DG_COMPLETE"

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

@@ -1,4 +1,3 @@
-import sys
 from collections.abc import Mapping
 from dataclasses import dataclass, replace
 from pathlib import Path
@@ -9,17 +8,17 @@
 from click.core import ParameterSource
 
 from dagster_dg.error import DgError, DgValidationError
-from dagster_dg.utils import get_toml_value
+from dagster_dg.utils import get_toml_value, is_macos, is_windows
 
 T = TypeVar("T")
 
 DEFAULT_BUILTIN_COMPONENT_LIB = "dagster_components"
 
 
 def _get_default_cache_dir() -> Path:
-    if sys.platform == "win32":
+    if is_windows():
         return Path.home() / "AppData" / "dg" / "cache"
-    elif sys.platform == "darwin":
+    elif is_macos():
         return Path.home() / "Library" / "Caches" / "dg"
     else:
         return Path.home() / ".cache" / "dg"

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

@@ -21,6 +21,7 @@
     get_path_for_module,
     get_path_for_package,
     get_uv_command_env,
+    get_venv_executable,
     is_executable_available,
     is_package_installed,
     pushd,
@@ -128,7 +129,7 @@ def has_cache(self) -> bool:
         return self._cache is not None
 
     def get_cache_key(self, data_type: CachableDataType) -> tuple[str, str, str]:
-        path_parts = [str(part) for part in self.root_path.parts if part != "/"]
+        path_parts = [str(part) for part in self.root_path.parts if part != self.root_path.anchor]
         paths_to_hash = [
             self.root_path / "uv.lock",
             *([self.components_lib_path] if self.is_component_library else []),
@@ -212,7 +213,7 @@ def code_location_python_executable(self) -> Path:
             raise DgError(
                 "`code_location_python_executable` is only available in a code location context"
             )
-        return self.root_path / ".venv" / "bin" / "python"
+        return self.root_path / get_venv_executable(Path(".venv"))
 
     @cached_property
     def components_package_name(self) -> str:

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

@@ -72,7 +72,7 @@ def get_pyproject_toml_dev_dependencies(use_editable_dagster: bool) -> str:
 
 def get_pyproject_toml_uv_sources(editable_dagster_root: Path) -> str:
     lib_lines = [
-        f'{path.name} = {{ path = "{path}", editable = true }}'
+        f"{path.name} = {{ path = '{path}', editable = true }}"
         for path in _gather_dagster_packages(editable_dagster_root)
     ]
     return "\n".join(
@@ -140,7 +140,7 @@ def scaffold_component_type(dg_context: DgContext, name: str) -> None:
     scaffold_subtree(
         path=root_path,
         name_placeholder="COMPONENT_TYPE_NAME_PLACEHOLDER",
-        templates_path=os.path.join(os.path.dirname(__file__), "templates", "COMPONENT_TYPE"),
+        templates_path=str(Path(__file__).parent / "templates" / "COMPONENT_TYPE"),
         project_name=name,
         component_type_class_name=camelcase(name),
         name=name,

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

@@ -2,9 +2,12 @@
 import importlib.util
 import json
 import os
+import platform
 import posixpath
 import re
 import shutil
+import signal
+import subprocess
 import sys
 import textwrap
 from collections.abc import Iterable, Iterator, Mapping, Sequence
@@ -30,6 +33,27 @@
 CLI_CONFIG_KEY = "config"
 
 
+def is_windows() -> bool:
+    return platform.system() == "Windows"
+
+
+def is_macos() -> bool:
+    return platform.system() == "Darwin"
+
+
+def get_venv_executable(venv_dir: Path) -> Path:
+    if is_windows():
+        return venv_dir / "Scripts" / "python.exe"
+    else:
+        return venv_dir / "bin" / "python"
+
+
+def install_to_venv(venv_dir: Path, install_args: list[str]) -> None:
+    executable = get_venv_executable(venv_dir)
+    command = ["uv", "pip", "install", "--python", str(executable), *install_args]
+    subprocess.run(command, check=True)
+
+
 # Temporarily places a path at the front of sys.path, ensuring that any modules in that path are
 # importable.
 @contextlib.contextmanager
@@ -84,6 +108,12 @@ def is_executable_available(command: str) -> bool:
     return bool(shutil.which(command))
 
 
+# Short for "normalize path"-- use this to get the platform-correct string representation of an
+# existing string path.
+def npath(path: str):
+    return str(Path(path))
+
+
 # uv commands should be executed in an environment with no pre-existing VIRTUAL_ENV set. If this
 # variable is set (common during development) and does not match the venv resolved by uv, it prints
 # undesireable warnings.
@@ -286,6 +316,11 @@ def exit_with_error(error_message: str) -> Never:
     sys.exit(1)
 
 
+# ########################
+# ##### ERROR MESSAGES
+# ########################
+
+
 def _format_error_message(message: str) -> str:
     # width=10000 unwraps any hardwrapping
     return textwrap.dedent(textwrap.fill(message, width=10000))
@@ -327,6 +362,38 @@ def generate_missing_component_type_error_message(component_key_str: str) -> str
 installed.
 """
 
+# ########################
+# ##### SUBPROCESSES
+# ########################
+
+# Windows subprocess termination utilities. See here for why we send CTRL_BREAK_EVENT on Windows:
+# https://stefan.sofa-rockers.org/2013/08/15/handling-sub-process-hierarchies-python-linux-os-x/
+
+
+def interrupt_subprocess(pid: int, interrupt_signal: Any = None) -> None:
+    """Send CTRL_BREAK_EVENT on Windows, SIGINT on other platforms."""
+    if is_windows():
+        # Pyright tries to be too smart and thinks that signal.CTRL_BREAK_EVENT is not valid because
+        # it doesn't realize OS is windows.
+        os.kill(pid, interrupt_signal or signal.CTRL_BREAK_EVENT)  # type: ignore
+    else:
+        os.kill(pid, interrupt_signal or signal.SIGINT)
+
+
+def open_subprocess(command: Sequence[str], **kwargs: Any) -> subprocess.Popen:
+    """Sets the correct flags to support graceful termination."""
+    creationflags = 0
+    if is_windows():
+        # Pyright tries to be too smart and thinks that signal.CTRL_BREAK_EVENT is not valid because
+        # it doesn't realize OS is windows.
+        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP  # type: ignore
+    return subprocess.Popen(
+        command,
+        creationflags=creationflags,
+        **kwargs,
+    )
+
+
 # ########################
 # ##### CUSTOM CLICK SUBCLASSES
 # ########################