[externals] JsonSchema

.buildkite/dagster-buildkite/dagster_buildkite/package_spec.py

@@ -1,6 +1,6 @@
 import logging
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Callable, List, Mapping, Optional, Union
 
@@ -24,8 +24,10 @@
 
 _CORE_PACKAGES = [
     "python_modules/dagster",
-    "python_modules/dagit",
+    "python_modules/dagster-externals",
     "python_modules/dagster-graphql",
+    "python_modules/dagster-webserver",
+    "python_modules/dagit",
     "js_modules/dagster-ui",
 ]
 
@@ -108,6 +110,8 @@ class PackageSpec:
         timeout_in_minutes (int, optional): Fail after this many minutes.
         queue (BuildkiteQueue, optional): Schedule steps to this queue.
         run_pytest (bool, optional): Whether to run pytest. Enabled by default.
+        other_tox_envs (List[str], optional): Other tox testenvs to run as part of the package step
+            group. Defaults to [].
     """
 
     directory: str
@@ -126,6 +130,7 @@ class PackageSpec:
     queue: Optional[BuildkiteQueue] = None
     run_pytest: bool = True
     always_run_if: Optional[Callable[[], bool]] = None
+    other_tox_envs: List[str] = field(default_factory=list)
 
     def __post_init__(self):
         if not self.name:
@@ -208,6 +213,9 @@ def build_steps(self) -> List[BuildkiteTopLevelStep]:
                         )
                     )
 
+        for testenv in self.other_tox_envs:
+            steps.append(build_tox_step(self.directory, testenv, base_label=testenv))
+
         emoji = _PACKAGE_TYPE_TO_EMOJI_MAP[self.package_type]  # type: ignore[index]
         if len(steps) >= 2:
             return [

.buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py

@@ -352,6 +352,7 @@ def k8s_extra_cmds(version: str, _) -> List[str]:
 
 LIBRARY_PACKAGES_WITH_CUSTOM_CONFIG: List[PackageSpec] = [
     PackageSpec("python_modules/automation"),
+    PackageSpec("python_modules/dagster-externals", other_tox_envs=["jsonschema"]),
     PackageSpec("python_modules/dagster-webserver", pytest_extra_cmds=ui_extra_cmds),
     PackageSpec(
         "python_modules/dagster",

Makefile

@@ -92,3 +92,6 @@ check_manifest:
 	check-manifest python_modules/dagster-webserver
 	check-manifest python_modules/dagster-graphql
 	ls python_modules/libraries | xargs -n 1 -Ipkg check-manifest python_modules/libraries/pkg
+
+externals_json_schema:
+	python scripts/generate_externals_json_schema.py

python_modules/dagster-externals/MANIFEST.in

@@ -1,3 +1,4 @@
 include README.md
 include LICENSE
-include dagster_external/py.typed
+include dagster_externals/py.typed
+include externals_protocol_schema.json

python_modules/dagster-externals/dagster_externals_tests/test_context.py

@@ -1,5 +1,8 @@
+import json
+import os
 from unittest.mock import MagicMock
 
+import jsonschema
 import pytest
 from dagster_externals._context import ExternalExecutionContext
 from dagster_externals._protocol import (
@@ -23,11 +26,18 @@
     extras={},
 )
 
+JSON_SCHEMA_PATH = os.path.join(os.path.dirname(__file__), "../externals_protocol_schema.json")
+
+with open(JSON_SCHEMA_PATH) as f:
+    JSON_SCHEMA = json.load(f)
+
 
 def _make_external_execution_context(**kwargs):
-    kwargs = {**TEST_EXTERNAL_EXECUTION_CONTEXT_DEFAULTS, **kwargs}
+    data = ExternalExecutionContextData(**{**TEST_EXTERNAL_EXECUTION_CONTEXT_DEFAULTS, **kwargs})
+    # This will error if the context doesn't match the schema
+    jsonschema.validate(data, JSON_SCHEMA)
     return ExternalExecutionContext(
-        data=ExternalExecutionContextData(**kwargs),
+        data=data,
         output_stream=MagicMock(),
     )
 
@@ -139,3 +149,13 @@ def test_extras_context():
     assert context.get_extra("foo") == "bar"
     with pytest.raises(DagsterExternalError, match="Extra `bar` is undefined"):
         context.get_extra("bar")
+
+
+def test_notification_json_schema_validation():
+    notification = {"method": "foo", "params": {"bar": "baz"}}
+    jsonschema.validate(notification, JSON_SCHEMA)
+
+
+def test_json_schema_rejects_invalid():
+    with pytest.raises(jsonschema.ValidationError):
+        jsonschema.validate({"foo": "bar"}, JSON_SCHEMA)