DagsterLibraryRegistry (#12266)

Add a means to track which dagster libraries have been loaded in a
process. This can be used for example to show which libraries and their
versions were used when loading definitions for a specific code
location.

### How I Tested These Changes

added test, can add more

```
(py39) ~/dagster:al/02-10-DagsterLibraryRegistry$ ls -l python_modules/libraries/ | grep -v total | wc -l
      43
(py39) ~/dagster:al/02-10-DagsterLibraryRegistry$ grep "DagsterLibraryRegistry.register" python_modules/libraries/ -r | wc -l
      43
```

remove a register call and ensure new test fails

python_modules/automation/automation_tests/test_repo.py

@@ -0,0 +1,17 @@
+import os
+import subprocess
+from pathlib import Path
+
+
+def test_all_libraries_register() -> None:
+    # attempt to ensure all libraries in the repository register with DagsterLibraryRegistry
+    register_call = "DagsterLibraryRegistry.register"
+
+    library_dir = Path(__file__).parents[2] / "libraries"
+    assert str(library_dir).endswith("python_modules/libraries")
+
+    for library in os.listdir(library_dir):
+        result = subprocess.run(["grep", register_call, (library_dir / library), "-r"])
+        assert (
+            result.returncode == 0
+        ), f"Dagster library {library} is missing call to {register_call}."

python_modules/dagit/dagit/__init__.py

@@ -1,5 +1,5 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .version import __version__
 
-check_dagster_package_version("dagit", __version__)
+DagsterLibraryRegistry.register("dagit", __version__)

python_modules/dagster-graphql/dagster_graphql/__init__.py

@@ -1,4 +1,4 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .client import (
     DagsterGraphQLClient as DagsterGraphQLClient,
@@ -11,4 +11,4 @@
 )
 from .version import __version__ as __version__
 
-check_dagster_package_version("dagster-graphql", __version__)
+DagsterLibraryRegistry.register("dagster-graphql", __version__)

python_modules/dagster/dagster/_core/libraries.py

@@ -0,0 +1,18 @@
+from typing import Dict, Mapping
+
+from dagster._core.utils import check_dagster_package_version
+
+from ..version import __version__
+
+
+class DagsterLibraryRegistry:
+    _libraries: Dict[str, str] = {"dagster": __version__}
+
+    @classmethod
+    def register(cls, name: str, version: str):
+        check_dagster_package_version(name, version)
+        cls._libraries[name] = version
+
+    @classmethod
+    def get(cls) -> Mapping[str, str]:
+        return cls._libraries.copy()

python_modules/dagster/dagster_tests/core_tests/test_utils.py

@@ -2,6 +2,7 @@
 
 import dagster.version
 import pytest
+from dagster._core.libraries import DagsterLibraryRegistry
 from dagster._core.test_utils import environ
 from dagster._core.utils import check_dagster_package_version, parse_env_var
 from dagster._utils import library_version_from_core_version
@@ -62,3 +63,7 @@ def test_library_version_from_core_version():
     assert library_version_from_core_version("1.1.16pre0") == "0.17.16rc0"
     assert library_version_from_core_version("1.1.16rc0") == "0.17.16rc0"
     assert library_version_from_core_version("1.1.16post0") == "0.17.16post0"
+
+
+def test_library_registry():
+    assert DagsterLibraryRegistry.get() == {"dagster": dagster.version.__version__}

python_modules/libraries/dagster-airbyte/dagster_airbyte/__init__.py

@@ -1,8 +1,6 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 try:
-    import dagster_managed_elements  # noqa: F401
-
     from .managed import (
         AirbyteConnection as AirbyteConnection,
         AirbyteDestination as AirbyteDestination,
@@ -30,4 +28,4 @@
 from .types import AirbyteOutput as AirbyteOutput
 from .version import __version__ as __version__
 
-check_dagster_package_version("dagster-airbyte", __version__)
+DagsterLibraryRegistry.register("dagster-airbyte", __version__)

python_modules/libraries/dagster-airflow/dagster_airflow/__init__.py

@@ -1,5 +1,5 @@
 from airflow.plugins_manager import AirflowPlugin
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .dagster_asset_factory import (
     load_assets_from_airflow_dag as load_assets_from_airflow_dag,
@@ -22,7 +22,7 @@
 from .resources import make_ephemeral_airflow_db_resource as make_ephemeral_airflow_db_resource
 from .version import __version__ as __version__
 
-check_dagster_package_version("dagster-airflow", __version__)
+DagsterLibraryRegistry.register("dagster-airflow", __version__)
 
 __all__ = [
     "make_dagster_definitions_from_airflow_dags_path",

python_modules/libraries/dagster-aws/dagster_aws/__init__.py

@@ -1,5 +1,5 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .version import __version__
 
-check_dagster_package_version("dagster-aws", __version__)
+DagsterLibraryRegistry.register("dagster-aws", __version__)

python_modules/libraries/dagster-azure/dagster_azure/__init__.py

@@ -1,5 +1,5 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .version import __version__
 
-check_dagster_package_version("dagster-azure", __version__)
+DagsterLibraryRegistry.register("dagster-azure", __version__)

python_modules/libraries/dagster-celery-docker/dagster_celery_docker/__init__.py

@@ -1,6 +1,6 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .executor import celery_docker_executor as celery_docker_executor
 from .version import __version__ as __version__
 
-check_dagster_package_version("dagster-celery-docker", __version__)
+DagsterLibraryRegistry.register("dagster-celery-docker", __version__)

python_modules/libraries/dagster-celery-k8s/dagster_celery_k8s/__init__.py

@@ -1,7 +1,7 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .executor import celery_k8s_job_executor as celery_k8s_job_executor
 from .launcher import CeleryK8sRunLauncher as CeleryK8sRunLauncher
 from .version import __version__ as __version__
 
-check_dagster_package_version("dagster-celery-k8s", __version__)
+DagsterLibraryRegistry.register("dagster-celery-k8s", __version__)

python_modules/libraries/dagster-celery/dagster_celery/__init__.py

@@ -1,8 +1,8 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .executor import celery_executor
 from .version import __version__
 
-check_dagster_package_version("dagster-celery", __version__)
+DagsterLibraryRegistry.register("dagster-celery", __version__)
 
 __all__ = ["celery_executor"]

python_modules/libraries/dagster-census/dagster_census/__init__.py

@@ -1,11 +1,11 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .ops import census_trigger_sync_op
 from .resources import CensusResource, census_resource
 from .types import CensusOutput
 from .version import __version__
 
-check_dagster_package_version("dagster-census", __version__)
+DagsterLibraryRegistry.register("dagster-census", __version__)
 
 __all__ = [
     "CensusResource",

python_modules/libraries/dagster-dask/dagster_dask/__init__.py

@@ -1,8 +1,8 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .data_frame import DataFrame as DataFrame
 from .executor import dask_executor as dask_executor
 from .resources import dask_resource as dask_resource
 from .version import __version__ as __version__
 
-check_dagster_package_version("dagster-dask", __version__)
+DagsterLibraryRegistry.register("dagster-dask", __version__)

python_modules/libraries/dagster-databricks/dagster_databricks/__init__.py

@@ -8,7 +8,7 @@
     to execute an arbitrary task in Databricks.
 """
 
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .databricks import DatabricksClient, DatabricksError, DatabricksJobRunner
 from .databricks_pyspark_step_launcher import (
@@ -25,7 +25,7 @@
 )
 from .version import __version__
 
-check_dagster_package_version("dagster-databricks", __version__)
+DagsterLibraryRegistry.register("dagster-databricks", __version__)
 
 __all__ = [
     "create_databricks_job_op",

python_modules/libraries/dagster-datadog/dagster_datadog/__init__.py

@@ -1,8 +1,8 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .resources import datadog_resource
 from .version import __version__
 
-check_dagster_package_version("dagster-datadog", __version__)
+DagsterLibraryRegistry.register("dagster-datadog", __version__)
 
 __all__ = ["datadog_resource"]

python_modules/libraries/dagster-datahub/dagster_datahub/__init__.py

@@ -1,8 +1,8 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .resources import datahub_kafka_emitter, datahub_rest_emitter
 from .version import __version__
 
-check_dagster_package_version("dagster-datahub", __version__)
+DagsterLibraryRegistry.register("dagster-datahub", __version__)
 
 __all__ = ["datahub_rest_emitter", "datahub_kafka_emitter"]

python_modules/libraries/dagster-dbt/dagster_dbt/__init__.py

@@ -1,4 +1,4 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .asset_defs import (
     load_assets_from_dbt_manifest as load_assets_from_dbt_manifest,
@@ -50,4 +50,4 @@
 from .types import DbtOutput as DbtOutput
 from .version import __version__ as __version__
 
-check_dagster_package_version("dagster-dbt", __version__)
+DagsterLibraryRegistry.register("dagster-dbt", __version__)

python_modules/libraries/dagster-docker/dagster_docker/__init__.py

@@ -1,4 +1,4 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .docker_executor import docker_executor as docker_executor
 from .docker_run_launcher import DockerRunLauncher as DockerRunLauncher
@@ -8,4 +8,4 @@
 )
 from .version import __version__
 
-check_dagster_package_version("dagster-docker", __version__)
+DagsterLibraryRegistry.register("dagster-docker", __version__)

python_modules/libraries/dagster-duckdb-pandas/dagster_duckdb_pandas/__init__.py

@@ -1,4 +1,9 @@
+from dagster._core.libraries import DagsterLibraryRegistry
+
 from .duckdb_pandas_type_handler import (
     DuckDBPandasTypeHandler as DuckDBPandasTypeHandler,
     duckdb_pandas_io_manager as duckdb_pandas_io_manager,
 )
+from .version import __version__
+
+DagsterLibraryRegistry.register("dagster-duckdb-pandas", __version__)

python_modules/libraries/dagster-duckdb-pyspark/dagster_duckdb_pyspark/__init__.py

@@ -1,4 +1,9 @@
+from dagster._core.libraries import DagsterLibraryRegistry
+
 from .duckdb_pyspark_type_handler import (
     DuckDBPySparkTypeHandler as DuckDBPySparkTypeHandler,
     duckdb_pyspark_io_manager as duckdb_pyspark_io_manager,
 )
+from .version import __version__
+
+DagsterLibraryRegistry.register("dagster-duckdb-pyspark", __version__)

python_modules/libraries/dagster-duckdb/dagster_duckdb/__init__.py

@@ -1 +1,6 @@
+from dagster._core.libraries import DagsterLibraryRegistry
+
 from .io_manager import build_duckdb_io_manager as build_duckdb_io_manager
+from .version import __version__
+
+DagsterLibraryRegistry.register("dagster-duckdb", __version__)

python_modules/libraries/dagster-fivetran/dagster_fivetran/__init__.py

@@ -1,4 +1,4 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .asset_defs import (
     build_fivetran_assets as build_fivetran_assets,
@@ -16,8 +16,6 @@
 from .version import __version__ as __version__
 
 try:
-    import dagster_managed_elements  # noqa: F401
-
     from .managed import (
         FivetranConnector as FivetranConnector,
         FivetranDestination as FivetranDestination,
@@ -28,4 +26,4 @@
     pass
 
 
-check_dagster_package_version("dagster-fivetran", __version__)
+DagsterLibraryRegistry.register("dagster-fivetran", __version__)

python_modules/libraries/dagster-gcp/dagster_gcp/__init__.py

@@ -1,4 +1,4 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .bigquery.ops import (
     bq_create_dataset,
@@ -15,7 +15,7 @@
 from .gcs import GCSFileHandle, gcs_file_manager, gcs_resource
 from .version import __version__
 
-check_dagster_package_version("dagster-gcp", __version__)
+DagsterLibraryRegistry.register("dagster-gcp", __version__)
 
 __all__ = [
     "BigQueryError",

python_modules/libraries/dagster-ge/dagster_ge/__init__.py

@@ -1,3 +1,8 @@
+from dagster._core.libraries import DagsterLibraryRegistry
+
 from .factory import ge_validation_op_factory
+from .version import __version__
+
+DagsterLibraryRegistry.register("dagster-ge", __version__)
 
 __all__ = ["ge_validation_op_factory"]

python_modules/libraries/dagster-github/dagster_github/__init__.py

@@ -1,8 +1,8 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .resources import github_resource
 from .version import __version__
 
-check_dagster_package_version("dagster-github", __version__)
+DagsterLibraryRegistry.register("dagster-github", __version__)
 
 __all__ = ["github_resource"]

python_modules/libraries/dagster-k8s/dagster_k8s/__init__.py

@@ -1,4 +1,4 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .executor import k8s_job_executor as k8s_job_executor
 from .job import (
@@ -12,4 +12,4 @@
 )
 from .version import __version__ as __version__
 
-check_dagster_package_version("dagster-k8s", __version__)
+DagsterLibraryRegistry.register("dagster-k8s", __version__)

python_modules/libraries/dagster-managed-elements/dagster_managed_elements/__init__.py

@@ -1 +1,6 @@
+from dagster._core.libraries import DagsterLibraryRegistry
+
 from .types import *  # noqa: F403
+from .version import __version__
+
+DagsterLibraryRegistry.register("dagster-managed-elements", __version__)

python_modules/libraries/dagster-mlflow/dagster_mlflow/__init__.py

@@ -1,9 +1,9 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .hooks import end_mlflow_on_run_finished
 from .resources import mlflow_tracking
 from .version import __version__
 
-check_dagster_package_version("dagster-mlflow", __version__)
+DagsterLibraryRegistry.register("dagster-mlflow", __version__)
 
 __all__ = ["mlflow_tracking", "end_mlflow_on_run_finished"]

python_modules/libraries/dagster-msteams/dagster_msteams/__init__.py

@@ -1,4 +1,4 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .card import Card as Card
 from .hooks import (
@@ -9,4 +9,4 @@
 from .sensors import make_teams_on_run_failure_sensor as make_teams_on_run_failure_sensor
 from .version import __version__ as __version__
 
-check_dagster_package_version("dagster-msteams", __version__)
+DagsterLibraryRegistry.register("dagster-msteams", __version__)

python_modules/libraries/dagster-mysql/dagster_mysql/__init__.py

@@ -1,10 +1,10 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .event_log import MySQLEventLogStorage
 from .run_storage import MySQLRunStorage
 from .schedule_storage import MySQLScheduleStorage
 from .storage import DagsterMySQLStorage
 from .version import __version__
 
-check_dagster_package_version("dagster-mysql", __version__)
+DagsterLibraryRegistry.register("dagster-mysql", __version__)
 __all__ = ["DagsterMySQLStorage", "MySQLEventLogStorage", "MySQLRunStorage", "MySQLScheduleStorage"]

python_modules/libraries/dagster-pagerduty/dagster_pagerduty/__init__.py

@@ -1,7 +1,7 @@
-from dagster._core.utils import check_dagster_package_version
+from dagster._core.libraries import DagsterLibraryRegistry
 
 from .hooks import pagerduty_on_failure as pagerduty_on_failure
 from .resources import pagerduty_resource as pagerduty_resource
 from .version import __version__ as __version__
 
-check_dagster_package_version("dagster-pagerduty", __version__)
+DagsterLibraryRegistry.register("dagster-pagerduty", __version__)