add CLI command to print the storage schema version (#7910)

* add CLI command to print the storage schema

* update instance info cli to show schema info

python_modules/dagster/dagster/cli/instance.py

@@ -32,8 +32,12 @@ def info_command():
 
         click.echo("$DAGSTER_HOME: {}\n".format(home))
 
+        click.echo("\nInstance configuration:\n-----------------------")
         click.echo(instance.info_str())
 
+        click.echo("\nStorage schema state:\n---------------------")
+        click.echo(instance.schema_str())
+
 
 @instance_cli.command(name="migrate", help="Automatically migrate an out of date instance.")
 def migrate_command():

python_modules/dagster/dagster/core/instance/__init__.py

@@ -530,6 +530,28 @@ def info_dict(self):
     def info_str(self) -> str:
         return yaml.dump(self.info_dict(), default_flow_style=False, sort_keys=False)
 
+    def schema_str(self) -> str:
+        def _schema_dict(alembic_version):
+            if not alembic_version:
+                return None
+            db_revision, head_revision = alembic_version
+            return {
+                "current": db_revision,
+                "latest": head_revision,
+            }
+
+        return yaml.dump(
+            {
+                "schema": {
+                    "event_log_storage": _schema_dict(self._event_storage.alembic_version()),
+                    "run_storage": _schema_dict(self._event_storage.alembic_version()),
+                    "schedule_storage": _schema_dict(self._event_storage.alembic_version()),
+                }
+            },
+            default_flow_style=False,
+            sort_keys=False,
+        )
+
     @property
     def run_storage(self) -> "RunStorage":
         return self._run_storage

python_modules/dagster/dagster/core/storage/event_log/base.py

@@ -328,6 +328,9 @@ def get_materialization_count_by_partition(
     ) -> Mapping[AssetKey, Mapping[str, int]]:
         pass
 
+    def alembic_version(self):
+        return None
+
 
 def extract_asset_events_cursor(cursor, before_cursor, after_cursor, ascending):
     if cursor:

python_modules/dagster/dagster/core/storage/event_log/sqlite/sqlite_event_log.py

@@ -404,6 +404,11 @@ def dispose(self):
             self._obs.stop()
             self._obs.join(timeout=15)
 
+    def alembic_version(self):
+        alembic_config = get_alembic_config(__file__)
+        with self.index_connection() as conn:
+            return check_alembic_revision(alembic_config, conn)
+
 
 class SqliteEventLogStorageWatchdog(PatternMatchingEventHandler):
     def __init__(self, event_log_storage, run_id, callback, start_cursor, **kwargs):

python_modules/dagster/dagster/core/storage/runs/base.py

@@ -384,3 +384,6 @@ def add_backfill(self, partition_backfill: PartitionBackfill):
     @abstractmethod
     def update_backfill(self, partition_backfill: PartitionBackfill):
         """Update a partition backfill in run storage"""
+
+    def alembic_version(self):
+        return None

python_modules/dagster/dagster/core/storage/runs/sqlite/sqlite_run_storage.py

@@ -159,3 +159,8 @@ def delete_run(self, run_id):
         with self.connect() as conn:
             conn.execute(remove_tags)
             conn.execute(remove_run)
+
+    def alembic_version(self):
+        alembic_config = get_alembic_config(__file__)
+        with self.connect() as conn:
+            return check_alembic_revision(alembic_config, conn)

python_modules/dagster/dagster/core/storage/schedules/base.py

@@ -130,3 +130,6 @@ def optimize(self, print_fn: Optional[Callable] = None, force_rebuild_all: bool
 
     def optimize_for_dagit(self, statement_timeout: int):
         """Allows for optimizing database connection / use in the context of a long lived dagit process"""
+
+    def alembic_version(self):
+        return None

python_modules/dagster/dagster/core/storage/schedules/sqlite/sqlite_schedule_storage.py

@@ -90,3 +90,8 @@ def upgrade(self):
         alembic_config = get_alembic_config(__file__)
         with self.connect() as conn:
             run_alembic_upgrade(alembic_config, conn)
+
+    def alembic_version(self):
+        alembic_config = get_alembic_config(__file__)
+        with self.connect() as conn:
+            return check_alembic_revision(alembic_config, conn)

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

@@ -8,8 +8,12 @@
     SqlPollingEventWatcher,
 )
 from dagster.core.storage.event_log.migration import ASSET_KEY_INDEX_COLS
-from dagster.core.storage.sql import stamp_alembic_rev  # pylint: disable=unused-import
-from dagster.core.storage.sql import create_engine, run_alembic_upgrade
+from dagster.core.storage.sql import (
+    check_alembic_revision,
+    create_engine,
+    run_alembic_upgrade,
+    stamp_alembic_rev,
+)
 from dagster.serdes import ConfigurableClass, ConfigurableClassData
 
 from ..utils import (
@@ -181,3 +185,8 @@ def dispose(self):
         if not self._disposed:
             self._disposed = True
             self._event_watcher.close()
+
+    def alembic_version(self):
+        alembic_config = mysql_alembic_config(__file__)
+        with self._connect() as conn:
+            return check_alembic_revision(alembic_config, conn)

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

@@ -7,8 +7,12 @@
     RunStorageSqlMetadata,
     SqlRunStorage,
 )
-from dagster.core.storage.sql import stamp_alembic_rev  # pylint: disable=unused-import
-from dagster.core.storage.sql import create_engine, run_alembic_upgrade
+from dagster.core.storage.sql import (
+    check_alembic_revision,
+    create_engine,
+    run_alembic_upgrade,
+    stamp_alembic_rev,
+)
 from dagster.serdes import ConfigurableClass, ConfigurableClassData, serialize_dagster_namedtuple
 from dagster.utils import utc_datetime_from_timestamp
 
@@ -144,3 +148,8 @@ def add_daemon_heartbeat(self, daemon_heartbeat):
                     body=serialize_dagster_namedtuple(daemon_heartbeat),
                 )
             )
+
+    def alembic_version(self):
+        alembic_config = mysql_alembic_config(__file__)
+        with self.connect() as conn:
+            return check_alembic_revision(alembic_config, conn)

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

@@ -4,7 +4,12 @@
 import dagster._check as check
 from dagster.core.storage.schedules import ScheduleStorageSqlMetadata, SqlScheduleStorage
 from dagster.core.storage.schedules.schema import InstigatorsTable
-from dagster.core.storage.sql import create_engine, run_alembic_upgrade, stamp_alembic_rev
+from dagster.core.storage.sql import (
+    check_alembic_revision,
+    create_engine,
+    run_alembic_upgrade,
+    stamp_alembic_rev,
+)
 from dagster.serdes import ConfigurableClass, ConfigurableClassData, serialize_dagster_namedtuple
 
 from ..utils import (
@@ -126,3 +131,8 @@ def _add_or_update_instigators_table(self, conn, state):
                 update_timestamp=pendulum.now("UTC"),
             )
         )
+
+    def alembic_version(self):
+        alembic_config = mysql_alembic_config(__file__)
+        with self.connect() as conn:
+            return check_alembic_revision(alembic_config, conn)

python_modules/libraries/dagster-postgres/dagster_postgres/event_log/event_log.py

@@ -11,7 +11,12 @@
     SqlEventLogStorageTable,
 )
 from dagster.core.storage.event_log.migration import ASSET_KEY_INDEX_COLS
-from dagster.core.storage.sql import create_engine, run_alembic_upgrade, stamp_alembic_rev
+from dagster.core.storage.sql import (
+    check_alembic_revision,
+    create_engine,
+    run_alembic_upgrade,
+    stamp_alembic_rev,
+)
 from dagster.serdes import ConfigurableClass, ConfigurableClassData, deserialize_as
 
 from ..utils import (
@@ -262,3 +267,8 @@ def dispose(self):
             self._disposed = True
             if self._event_watcher:
                 self._event_watcher.close()
+
+    def alembic_version(self):
+        alembic_config = pg_alembic_config(__file__)
+        with self._connect() as conn:
+            return check_alembic_revision(alembic_config, conn)

python_modules/libraries/dagster-postgres/dagster_postgres/run_storage/run_storage.py

@@ -7,7 +7,12 @@
     RunStorageSqlMetadata,
     SqlRunStorage,
 )
-from dagster.core.storage.sql import create_engine, run_alembic_upgrade, stamp_alembic_rev
+from dagster.core.storage.sql import (
+    check_alembic_revision,
+    create_engine,
+    run_alembic_upgrade,
+    stamp_alembic_rev,
+)
 from dagster.serdes import ConfigurableClass, ConfigurableClassData, serialize_dagster_namedtuple
 from dagster.utils import utc_datetime_from_timestamp
 
@@ -157,3 +162,8 @@ def add_daemon_heartbeat(self, daemon_heartbeat):
                     },
                 )
             )
+
+    def alembic_version(self):
+        alembic_config = pg_alembic_config(__file__)
+        with self.connect() as conn:
+            return check_alembic_revision(alembic_config, conn)

python_modules/libraries/dagster-postgres/dagster_postgres/schedule_storage/schedule_storage.py

@@ -4,7 +4,12 @@
 import dagster._check as check
 from dagster.core.storage.schedules import ScheduleStorageSqlMetadata, SqlScheduleStorage
 from dagster.core.storage.schedules.schema import InstigatorsTable
-from dagster.core.storage.sql import create_engine, run_alembic_upgrade, stamp_alembic_rev
+from dagster.core.storage.sql import (
+    check_alembic_revision,
+    create_engine,
+    run_alembic_upgrade,
+    stamp_alembic_rev,
+)
 from dagster.serdes import ConfigurableClass, ConfigurableClassData, serialize_dagster_namedtuple
 
 from ..utils import (
@@ -134,3 +139,8 @@ def _add_or_update_instigators_table(self, conn, state):
                 },
             )
         )
+
+    def alembic_version(self):
+        alembic_config = pg_alembic_config(__file__)
+        with self.connect() as conn:
+            return check_alembic_revision(alembic_config, conn)