[graphql] add RepositoryLocation.dagsterLibraryVersions (#12268)

expose over graphql the dagster libaries loaded by a code location

### How I Tested These Changes

added test

python_modules/dagster-graphql/dagster_graphql/schema/external.py

@@ -49,6 +49,14 @@
 GrapheneLocationStateChangeEventType = graphene.Enum.from_enum(LocationStateChangeEventType)
 
 
+class GrapheneDagsterLibraryVersion(graphene.ObjectType):
+    name = graphene.NonNull(graphene.String)
+    version = graphene.NonNull(graphene.String)
+
+    class Meta:
+        name = "DagsterLibraryVersion"
+
+
 class GrapheneRepositoryLocationLoadStatus(graphene.Enum):
     LOADING = "LOADING"
     LOADED = "LOADED"
@@ -74,11 +82,12 @@ class GrapheneRepositoryLocation(graphene.ObjectType):
     environment_path = graphene.String()
     repositories = non_null_list(lambda: GrapheneRepository)
     server_id = graphene.String()
+    dagsterLibraryVersions = graphene.List(graphene.NonNull(GrapheneDagsterLibraryVersion))
 
     class Meta:
         name = "RepositoryLocation"
 
-    def __init__(self, location):
+    def __init__(self, location: RepositoryLocation):
         self._location = check.inst_param(location, "location", RepositoryLocation)
         environment_path = (
             location.origin.loadable_target_origin.executable_path
@@ -99,7 +108,7 @@ def __init__(self, location):
             server_id=server_id,
         )
 
-    def resolve_id(self, _):
+    def resolve_id(self, _) -> str:
         return self.name
 
     def resolve_repositories(self, graphene_info: ResolveInfo):
@@ -108,6 +117,13 @@ def resolve_repositories(self, graphene_info: ResolveInfo):
             for repository in self._location.get_repositories().values()
         ]
 
+    def resolve_dagsterLibraryVersions(self, _: ResolveInfo):
+        libs = self._location.get_dagster_library_versions()
+        if libs is None:
+            return None
+
+        return [GrapheneDagsterLibraryVersion(name, ver) for name, ver in libs.items()]
+
 
 class GrapheneRepositoryLocationOrLoadError(graphene.Union):
     class Meta:

python_modules/dagster-graphql/dagster_graphql_tests/graphql/test_workspace.py

@@ -7,7 +7,9 @@
 from dagster._core.host_representation import ManagedGrpcPythonEnvRepositoryLocationOrigin
 from dagster._core.types.loadable_target_origin import LoadableTargetOrigin
 from dagster._core.workspace.load import location_origins_from_yaml_paths
+from dagster.version import __version__ as dagster_version
 from dagster_graphql.test.utils import execute_dagster_graphql
+from dagster_graphql.version import __version__ as dagster_graphql_version
 
 from .graphql_context_test_suite import GraphQLContextVariant, make_graphql_context_test_suite
 
@@ -29,6 +31,10 @@
                     name
                 }
                 isReloadSupported
+                dagsterLibraryVersions {
+                  name
+                  version
+                }
             }
             ... on PythonError {
               message
@@ -115,11 +121,15 @@ def test_load_workspace(self, graphql_context):
             )
 
             success_nodes = [
-                node
+                node["locationOrLoadError"]
                 for node in nodes
                 if node["locationOrLoadError"]["__typename"] == "RepositoryLocation"
             ]
             assert len(success_nodes) == 2
+            assert success_nodes[0]["dagsterLibraryVersions"] == [
+                {"name": "dagster", "version": dagster_version},
+                {"name": "dagster-graphql", "version": dagster_graphql_version},
+            ]
 
             failures = [
                 node for node in nodes if node["locationOrLoadError"]["__typename"] == "PythonError"

python_modules/dagster/dagster/_core/host_representation/repository_location.py

@@ -46,6 +46,7 @@
     RepositoryLocationOrigin,
 )
 from dagster._core.instance import DagsterInstance
+from dagster._core.libraries import DagsterLibraryRegistry
 from dagster._core.origin import RepositoryPythonOrigin
 from dagster._core.snap.execution_plan_snapshot import snapshot_from_execution_plan
 from dagster._grpc.impl import (
@@ -282,6 +283,10 @@ def get_repository_python_origin(self, repository_name: str) -> "RepositoryPytho
             container_context=self.container_context,
         )
 
+    @abstractmethod
+    def get_dagster_library_versions(self) -> Optional[Mapping[str, str]]:
+        ...
+
 
 class InProcessRepositoryLocation(RepositoryLocation):
     def __init__(self, origin: InProcessRepositoryLocationOrigin):
@@ -536,6 +541,9 @@ def get_external_notebook_data(self, notebook_path: str) -> bytes:
         check.str_param(notebook_path, "notebook_path")
         return get_notebook_data(notebook_path)
 
+    def get_dagster_library_versions(self) -> Mapping[str, str]:
+        return DagsterLibraryRegistry.get()
+
 
 class GrpcServerRepositoryLocation(RepositoryLocation):
     def __init__(
@@ -618,7 +626,7 @@ def __init__(
                 list_repositories_response.repository_code_pointer_dict
             )
             self._entry_point = list_repositories_response.entry_point
-
+            self._dagster_library_versions = list_repositories_response.dagster_library_versions
             self._container_image = (
                 list_repositories_response.container_image
                 or self._reload_current_image()  # Back-compat for older gRPC servers that did not include container_image in ListRepositoriesResponse
@@ -878,3 +886,6 @@ def get_external_partition_set_execution_param_data(
     def get_external_notebook_data(self, notebook_path: str) -> bytes:
         check.str_param(notebook_path, "notebook_path")
         return sync_get_streaming_external_notebook_data_grpc(self.client, notebook_path)
+
+    def get_dagster_library_versions(self) -> Optional[Mapping[str, str]]:
+        return self._dagster_library_versions