Add "--check-heads" option to "current" command

Fixes: #1705

diff --git a/alembic/command.py b/alembic/command.py
index 8e4854744ab32cd89f06c870a62a2cc711543c15..5fdbd03436b55c5843e5d4c85b8dd1993eb9c370 100644 (file)
--- a/alembic/command.py
+++ b/alembic/command.py
@@ -681,11 +681,16 @@ def branches(config: Config, verbose: bool = False) -> None:
             )
 
 
-def current(config: Config, verbose: bool = False) -> None:
+def current(
+    config: Config, check_heads: bool = False, verbose: bool = False
+) -> None:
     """Display the current revision for a database.
 
     :param config: a :class:`.Config` instance.
 
+    :param check_heads: Check if all head revisions are applied to the
+        database.  Exit with an error if this is not the case.
+
     :param verbose: output in verbose mode.
 
     """
@@ -698,6 +703,10 @@ def current(config: Config, verbose: bool = False) -> None:
                 "Current revision(s) for %s:",
                 util.obfuscate_url_pw(context.connection.engine.url),
             )
+        if check_heads and (
+            set(context.get_current_heads()) != set(script.get_heads())
+        ):
+            raise util.CommandError("Database is not on all head revisions")
         for rev in script.get_all_current(rev):
             config.print_stdout(rev.cmd_format(verbose))
 

diff --git a/alembic/config.py b/alembic/config.py
index b8c60a48971abf915549f4ac8b2a47757e9ef985..ad61b65888fab32932306ef7685fd7b01b873932 100644 (file)
--- a/alembic/config.py
+++ b/alembic/config.py
@@ -808,6 +808,17 @@ class CommandLine:
                 "environment and version locations",
             ),
         ),
+        "check_heads": (
+            "-c",
+            "--check-if-heads",
+            dict(
+                action="store_true",
+                help=(
+                    "Check if all head revisions are applied to the database. "
+                    "Exit with an error code if this is not the case."
+                ),
+            ),
+        ),
     }
     _POSITIONAL_OPTS = {
         "directory": dict(help="location of scripts directory"),

diff --git a/docs/build/changelog.rst b/docs/build/changelog.rst
index f411078d5c70327a420a3fde3b576f245189070f..286c5ddbfeba4edb82a2af11517316ed6dddcd0f 100644 (file)
--- a/docs/build/changelog.rst
+++ b/docs/build/changelog.rst
@@ -7,6 +7,15 @@ Changelog
     :version: 1.17.1
     :include_notes_from: unreleased
 
+    .. change::
+        :tags: usecase, commands
+        :tickets: 1705
+
+        Added ``--check-heads`` option to ``current`` command which
+        checks if all head revisions are applied to the database.
+        The command exists with a non-zero exit code if this is not the
+        case.
+
 .. changelog::
     :version: 1.17.0
     :released: October 11, 2025

diff --git a/tests/test_command.py b/tests/test_command.py
index 5920ac706ec9bd149fd29261093e53d1b981e712..603cef45659155763a999f8655403db3a46771dc 100644 (file)
--- a/tests/test_command.py
+++ b/tests/test_command.py
@@ -10,6 +10,7 @@ import re
 import shutil
 from typing import cast
 
+import pytest
 from sqlalchemy import exc as sqla_exc
 from sqlalchemy import text
 from sqlalchemy import VARCHAR
@@ -345,6 +346,35 @@ class CurrentTest(_BufMixin, TestBase):
         with self._assert_lines(["a2", "b3"]):
             command.current(self.cfg)
 
+    def test_check_if_head_success(self):
+        """
+        "--check-if-head" succeeds if all head revisions are applied.
+        """
+        command.stamp(self.cfg, ())
+        command.stamp(self.cfg, (self.a3.revision, self.b3.revision))
+        with self._assert_lines(["a3", "b3"]):
+            command.current(self.cfg, check_heads=True)
+
+    @pytest.mark.parametrize(
+        "revs", [("a2",), ("a3",), ("b3",), ("a2", "b3"), ("a3", "b2")]
+    )
+    def test_check_if_head_fail(self, revs):
+        """
+        "--check-if-head" succeeds if all head revisions are applied.
+        """
+        command.stamp(self.cfg, ())
+        command.stamp(
+            self.cfg, tuple(getattr(self, rev).revision for rev in revs)
+        )
+        assert_raises_message(
+            util.CommandError,
+            "Database is not on all head revisions",
+            command.current,
+            self.cfg,
+            check_heads=True,
+        )
+        command.stamp(self.cfg, ())
+
 
 class RevisionTest(TestBase):
     def setUp(self):