[dg] Serialize component type info with code location, add GQL to power in-app docs (#29167)

## Summary

Packages up component type JSON into the repository metadata of
`load_defs`, which can be accessed via GraphQL through the new
`locationDocsJson` query, which returns contents identical to the
generated JSON which powers `dg docs serve`.

Author: benpankow

Diff for: js_modules/dagster-ui/packages/ui-core/src/graphql/possibleTypes.generated.json

@@ -58,3 +58,16 @@ class GrapheneEnvVarWithConsumersListOrError(graphene.Union):
     class Meta:
         types = (GrapheneEnvVarWithConsumersList, GraphenePythonError)
         name = "EnvVarWithConsumersOrError"
+
+
+class GrapheneLocationDocsJson(graphene.ObjectType):
+    class Meta:
+        name = "LocationDocsJson"
+
+    json = graphene.NonNull(graphene.JSONString)
+
+
+class GrapheneLocationDocsJsonOrError(graphene.Union):
+    class Meta:
+        types = (GrapheneLocationDocsJson, GraphenePythonError)
+        name = "LocationDocsJsonOrError"

Diff for: js_modules/dagster-ui/packages/ui-core/src/graphql/schema.graphql

@@ -21,6 +21,7 @@
 from dagster._core.scheduler.instigation import InstigatorStatus, InstigatorType
 from dagster._core.storage.event_log.base import AssetRecord
 from dagster._core.workspace.permissions import Permissions
+from dagster.components.core.load_defs import PLUGIN_COMPONENT_TYPES_JSON_METADATA_KEY
 
 from dagster_graphql.implementation.execution.backfill import get_asset_backfill_preview
 from dagster_graphql.implementation.external import (
@@ -118,7 +119,11 @@
     GraphenePartitionBackfillsOrError,
 )
 from dagster_graphql.schema.entity_key import GrapheneAssetKey
-from dagster_graphql.schema.env_vars import GrapheneEnvVarWithConsumersListOrError
+from dagster_graphql.schema.env_vars import (
+    GrapheneEnvVarWithConsumersListOrError,
+    GrapheneLocationDocsJson,
+    GrapheneLocationDocsJsonOrError,
+)
 from dagster_graphql.schema.external import (
     GrapheneRepositoriesOrError,
     GrapheneRepositoryConnection,
@@ -312,6 +317,12 @@ class Meta:
         description="Retrieve all the utilized environment variables for the given repo.",
     )
 
+    locationDocsJsonOrError = graphene.Field(
+        graphene.NonNull(GrapheneLocationDocsJsonOrError),
+        repositorySelector=graphene.NonNull(GrapheneRepositorySelector),
+        description="Retrieves JSON blob to drive integrated code location docs.",
+    )
+
     sensorOrError = graphene.Field(
         graphene.NonNull(GrapheneSensorOrError),
         sensorSelector=graphene.NonNull(GrapheneSensorSelector),
@@ -767,6 +778,27 @@ def resolve_utilizedEnvVarsOrError(self, graphene_info: ResolveInfo, **kwargs):
             RepositorySelector.from_graphql_input(kwargs.get("repositorySelector")),
         )
 
+    @capture_error
+    def resolve_locationDocsJsonOrError(
+        self, graphene_info: ResolveInfo, repositorySelector: GrapheneRepositorySelector
+    ) -> GrapheneLocationDocsJson:
+        repo_selector = RepositorySelector.from_graphql_input(repositorySelector)
+
+        location = graphene_info.context.get_code_location(repo_selector.location_name)
+        repository = location.get_repository(repo_selector.repository_name)
+        plugin_docs_json = (
+            cast(
+                list,
+                repository.repository_snap.metadata.get(
+                    PLUGIN_COMPONENT_TYPES_JSON_METADATA_KEY, [[]]
+                ),
+            )[0]
+            if repository.repository_snap.metadata
+            else []
+        )
+
+        return GrapheneLocationDocsJson(json=plugin_docs_json)
+
     @capture_error
     def resolve_sensorOrError(
         self, graphene_info: ResolveInfo, sensorSelector: GrapheneRepositorySelector

Diff for: js_modules/dagster-ui/packages/ui-core/src/graphql/types.ts

@@ -0,0 +1,82 @@
+import json
+from unittest import mock
+
+from dagster import Definitions
+from dagster._core.definitions.repository_definition.repository_definition import (
+    RepositoryDefinition,
+)
+from dagster._core.test_utils import ensure_dagster_tests_import, instance_for_test
+from dagster_graphql.test.utils import define_out_of_process_context, execute_dagster_graphql
+from dagster_shared.serdes.objects.package_entry import PluginObjectKey
+
+ensure_dagster_tests_import()
+
+GET_DOCS_JSON_QUERY = """
+query GetDocsJson {
+  locationDocsJsonOrError(repositorySelector: {repositoryLocationName: "test_location", repositoryName: "__repository__"}) {
+    __typename
+    ... on LocationDocsJson {
+      json
+    }
+    ... on PythonError {
+      message
+    }
+  }
+}
+"""
+
+
+def get_empty_repo() -> RepositoryDefinition:
+    return Definitions().get_repository_def()
+
+
+def test_get_empty_docs_json():
+    with (
+        instance_for_test() as instance,
+        define_out_of_process_context(__file__, "get_empty_repo", instance) as context,
+    ):
+        get_docs_json_result = execute_dagster_graphql(context, GET_DOCS_JSON_QUERY)
+        assert (
+            get_docs_json_result.data["locationDocsJsonOrError"]["__typename"] == "LocationDocsJson"
+        )
+        assert get_docs_json_result.data["locationDocsJsonOrError"]["json"] is not None
+
+        json_contents = json.loads(get_docs_json_result.data["locationDocsJsonOrError"]["json"])
+        assert len(json_contents) == 0
+
+
+def get_components_repo() -> RepositoryDefinition:
+    with mock.patch(
+        "dagster.components.core.package_entry.discover_entry_point_package_objects"
+    ) as mock_discover_entry_point_package_objects:
+        import dagster_test.components
+        from dagster.components.core.package_entry import get_package_objects_in_module
+
+        from dagster_graphql_tests.graphql.components import defs as defs
+
+        objects = {}
+        for name, obj in get_package_objects_in_module(dagster_test.components):
+            key = PluginObjectKey(name=name, namespace="dagster_test")
+            objects[key] = obj
+
+        mock_discover_entry_point_package_objects.return_value = objects
+
+        from dagster.components.core.load_defs import load_defs
+
+        return load_defs(defs).get_repository_def()
+
+
+def test_get_docs_json():
+    with (
+        instance_for_test() as instance,
+        define_out_of_process_context(__file__, "get_components_repo", instance) as context,
+    ):
+        get_docs_json_result = execute_dagster_graphql(context, GET_DOCS_JSON_QUERY)
+        assert (
+            get_docs_json_result.data["locationDocsJsonOrError"]["__typename"] == "LocationDocsJson"
+        )
+        assert get_docs_json_result.data["locationDocsJsonOrError"]["json"] is not None
+
+        json_contents = json.loads(get_docs_json_result.data["locationDocsJsonOrError"]["json"])
+        assert len(json_contents) == 1
+        assert json_contents[0]["name"] == "dagster_test"

Diff for: python_modules/dagster-graphql/dagster_graphql/schema/env_vars.py

@@ -40,5 +40,6 @@ def get_version() -> str:
         "requests",
         "starlette",  # used for run_in_threadpool utility fn
     ],
+    extras_require={"test": ["dagster-test"]},
     entry_points={"console_scripts": ["dagster-graphql = dagster_graphql.cli:main"]},
 )

Diff for: python_modules/dagster-graphql/dagster_graphql/schema/roots/query.py

@@ -17,7 +17,8 @@ deps =
   -e ../dagster-pipes
   -e ../libraries/dagster-shared
   postgres: -e ../libraries/dagster-postgres
-  -e .
+  -e ../dagster-test
+  -e .[test]
 allowlist_externals =
   /bin/bash
   uv

Diff for: python_modules/dagster-graphql/dagster_graphql_tests/graphql/components/__init__.py

@@ -2,12 +2,16 @@
 from pathlib import Path
 from types import ModuleType
 
+from dagster_shared.serdes.objects.package_entry import json_for_all_components
+
 from dagster._annotations import deprecated
 from dagster._core.definitions.definitions_class import Definitions
 from dagster._core.errors import DagsterInvalidDefinitionError
 from dagster._utils.warnings import suppress_dagster_warnings
 from dagster.components.core.context import ComponentLoadContext, use_component_load_context
 
+PLUGIN_COMPONENT_TYPES_JSON_METADATA_KEY = "plugin_component_types_json"
+
 
 @deprecated(breaking_version="0.2.0")
 @suppress_dagster_warnings
@@ -36,12 +40,21 @@ def load_defs(defs_root: ModuleType) -> Definitions:
             to apply to the definitions.
     """
     from dagster.components.core.defs_module import DefsModuleComponent
+    from dagster.components.core.package_entry import discover_entry_point_package_objects
+    from dagster.components.core.snapshot import get_package_entry_snap
 
     # create a top-level DefsModule component from the root module
     context = ComponentLoadContext.for_module(defs_root)
     root_component = DefsModuleComponent.from_context(context)
     if root_component is None:
         raise DagsterInvalidDefinitionError("Could not resolve root module to a component.")
 
+    library_objects = discover_entry_point_package_objects()
+    snaps = [get_package_entry_snap(key, obj) for key, obj in library_objects.items()]
+    components_json = json_for_all_components(snaps)
+
     with use_component_load_context(context):
-        return root_component.build_defs(context)
+        return Definitions.merge(
+            root_component.build_defs(context),
+            Definitions(metadata={PLUGIN_COMPONENT_TYPES_JSON_METADATA_KEY: components_json}),
+        )

Diff for: python_modules/dagster-graphql/dagster_graphql_tests/graphql/components/defs/__init__.py

@@ -8,13 +8,13 @@
 from typing import Optional
 
 import click
+from dagster_shared.serdes.objects.package_entry import json_for_all_components
 from yaspin import yaspin
 
 from dagster_dg.cli.shared_options import dg_global_options
 from dagster_dg.component import PluginObjectKey, RemotePluginRegistry
 from dagster_dg.config import normalize_cli_config
 from dagster_dg.context import DgContext
-from dagster_dg.docs import json_for_all_components
 from dagster_dg.utils import DgClickCommand, DgClickGroup, exit_with_error, pushd
 from dagster_dg.utils.telemetry import cli_telemetry_wrapper
 
@@ -27,7 +27,9 @@
     / "dg-docs-site"
 )
 DOCS_DIR = Path(__file__).parent.parent / "docs" / "packages" / "dg-docs-site"
-ACTIVE_DOCS_DIR = DOCS_DIR if DOCS_DIR.exists() else DEV_DOCS_DIR
+ACTIVE_DOCS_DIR = (
+    DOCS_DIR if DOCS_DIR.exists() and (DOCS_DIR / "package.json").exists() else DEV_DOCS_DIR
+)
 
 DOCS_JSON_PATH = ACTIVE_DOCS_DIR / "contents" / "generated.json"
 
@@ -65,7 +67,9 @@ def serve_docs_command(
 
     with pushd(ACTIVE_DOCS_DIR):
         DOCS_JSON_PATH.parent.mkdir(parents=True, exist_ok=True)
-        DOCS_JSON_PATH.write_text(json.dumps(json_for_all_components(registry), indent=2))
+        DOCS_JSON_PATH.write_text(
+            json.dumps(json_for_all_components([v for _, v in registry.items()]), indent=2)
+        )
         with yaspin(text="Verifying docs dependencies", color="blue") as spinner:
             yes = subprocess.Popen(["yes", "y"], stdout=subprocess.PIPE)
             try:
@@ -122,7 +126,9 @@ def build_docs_command(
 
     with pushd(ACTIVE_DOCS_DIR):
         DOCS_JSON_PATH.parent.mkdir(parents=True, exist_ok=True)
-        DOCS_JSON_PATH.write_text(json.dumps(json_for_all_components(registry), indent=2))
+        DOCS_JSON_PATH.write_text(
+            json.dumps(json_for_all_components([v for _, v in registry.items()]), indent=2)
+        )
         with yaspin(text="Verifying docs dependencies", color="blue") as spinner:
             yes = subprocess.Popen(["yes", "y"], stdout=subprocess.PIPE)
             try:
@@ -132,6 +138,7 @@ def build_docs_command(
             spinner.ok("✓")
 
         spinner = yaspin(text="Building docs", color="blue")
+        spinner.start()
         subprocess.check_output(["yarn", "build"])
         spinner.ok("✓")