CI: integrate alembic check

Author: guan404ming

.github/actions/migration_tests/action.yml

@@ -89,6 +89,15 @@ runs:
         COMPOSE_PROJECT_NAME: "docker-compose"
         AIRFLOW__DATABASE__EXTERNAL_DB_MANAGERS: "airflow.providers.fab.auth_manager.models.db.FABDBManager"
       if: env.BACKEND != 'sqlite'
+    - name: "Test model changes: ${{env.BACKEND}}"
+      shell: bash
+      run: >
+        breeze shell "airflow db reset -y &&
+        airflow db migrate --to-revision heads &&
+        airflow db check-models"
+      env:
+        COMPOSE_PROJECT_NAME: "docker-compose"
+        AIRFLOW__DATABASE__EXTERNAL_DB_MANAGERS: "airflow.providers.fab.auth_manager.models.db.FABDBManager"
     - name: "Bring any containers left down"
       shell: bash
       run: breeze down

airflow-core/src/airflow/cli/cli_config.py

@@ -1406,6 +1406,13 @@ class GroupCommand(NamedTuple):
         func=lazy_load_command("airflow.cli.commands.db_command.check"),
         args=(ARG_VERBOSE, ARG_DB_RETRY, ARG_DB_RETRY_DELAY),
     ),
+    ActionCommand(
+        name="check-models",
+        help="Check if there are model changes without a corresponding migration",
+        description="Check if the current models require new migrations to be generated",
+        func=lazy_load_command("airflow.cli.commands.db_command.check_models"),
+        args=(ARG_VERBOSE,),
+    ),
     ActionCommand(
         name="clean",
         help="Purge old records in metastore tables",

airflow-core/src/airflow/cli/commands/db_command.py

@@ -24,11 +24,13 @@
 from tempfile import NamedTemporaryFile
 from typing import TYPE_CHECKING
 
+from alembic import command
 from packaging.version import InvalidVersion, parse as parse_version
 from tenacity import Retrying, stop_after_attempt, wait_fixed
 
 from airflow import settings
 from airflow.exceptions import AirflowException
+from airflow.models import import_all_models
 from airflow.utils import cli as cli_utils, db
 from airflow.utils.db import _REVISION_HEADS_MAP
 from airflow.utils.db_cleanup import config_dict, drop_archived_tables, export_archived_records, run_cleanup
@@ -268,6 +270,21 @@ def _warn_remaining_retries(retrystate: RetryCallState):
             db.check()
 
 
+@cli_utils.action_cli(check_db=False)
+@providers_configuration_loaded
+def check_models(args):
+    """Check if there are model changes without a corresponding migration."""
+    import_all_models()
+    config = db._get_alembic_config()
+
+    try:
+        command.check(config)
+        print("Success: No model changes detected without migrations.")
+    except Exception as e:
+        print(f"Error: Model changes detected without migrations: {e}")
+        raise SystemExit(1)
+
+
 # lazily imported by CLI parser for `help` command
 all_tables = sorted(config_dict)
 

airflow-core/tests/unit/cli/commands/test_db_command.py

@@ -53,6 +53,12 @@ def test_cli_check_migrations(self, mock_wait_for_migrations):
 
         mock_wait_for_migrations.assert_called_once_with(timeout=60)
 
+    @mock.patch("airflow.cli.commands.db_command.command.check")
+    def test_cli_check_models(self, mock_check):
+        db_command.check_models(self.parser.parse_args(["db", "check-models"]))
+
+        mock_check.assert_called_once_with(timeout=60)
+
     @pytest.mark.parametrize(
         "args, called_with",
         [