Add db status command (#5560)

* add db status command

* fix status command

---------

Co-authored-by: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>

Author: Lendemor

reflex/model.py

@@ -19,6 +19,7 @@
 import sqlalchemy.ext.asyncio
 import sqlalchemy.orm
 from alembic.runtime.migration import MigrationContext
+from alembic.script.base import Script
 
 from reflex.base import Base
 from reflex.config import get_config
@@ -34,6 +35,41 @@
 from sqlmodel.ext.asyncio.session import AsyncSession  # noqa: E402
 
 
+def format_revision(
+    rev: Script,
+    current_rev: str | None,
+    current_reached_ref: list[bool],
+) -> str:
+    """Format a single revision for display.
+
+    Args:
+        rev: The alembic script object
+        current_rev: The currently applied revision ID
+        current_reached_ref: Mutable reference to track if we've reached current revision
+
+    Returns:
+        Formatted string for display
+    """
+    current = rev.revision
+    message = rev.doc
+
+    # Determine if this migration is applied
+    if current_rev is None:
+        is_applied = False
+    elif current == current_rev:
+        is_applied = True
+        current_reached_ref[0] = True
+    else:
+        is_applied = not current_reached_ref[0]
+
+    # Show checkmark or X with colors
+    status_icon = "[green]✓[/green]" if is_applied else "[red]✗[/red]"
+    head_marker = " (head)" if rev.is_head else ""
+
+    # Format output with message
+    return f"  [{status_icon}] {current}{head_marker}, {message}"
+
+
 def _safe_db_url_for_logging(url: str) -> str:
     """Remove username and password from the database URL for logging.
 
@@ -361,6 +397,25 @@ def alembic_init(cls):
             directory=str(environment.ALEMBIC_CONFIG.get().parent / "alembic"),
         )
 
+    @classmethod
+    def get_migration_history(cls):
+        """Get migration history with current database state.
+
+        Returns:
+            tuple: (current_revision, revisions_list) where revisions_list is in chronological order
+        """
+        # Get current revision from database
+        with cls.get_db_engine().connect() as connection:
+            context = MigrationContext.configure(connection)
+            current_rev = context.get_current_revision()
+
+        # Get all revisions from base to head
+        _, script_dir = cls._alembic_config()
+        revisions = list(script_dir.walk_revisions())
+        revisions.reverse()  # Reverse to get chronological order (base first)
+
+        return current_rev, revisions
+
     @classmethod
     def alembic_autogenerate(
         cls,

reflex/reflex.py

@@ -540,6 +540,39 @@ def migrate():
     prerequisites.check_schema_up_to_date()
 
 
+@db_cli.command()
+def status():
+    """Check the status of the database schema."""
+    from reflex.model import Model, format_revision
+    from reflex.utils import prerequisites
+
+    prerequisites.get_app()
+    if not prerequisites.check_db_initialized():
+        console.info(
+            "Database is not initialized. Run [bold]reflex db init[/bold] to initialize."
+        )
+        return
+
+    # Run alembic check command and display output
+    import reflex.config
+
+    config = reflex.config.get_config()
+    console.print(f"[bold]\\[{config.db_url}][/bold]")
+
+    # Get migration history using Model method
+    current_rev, revisions = Model.get_migration_history()
+    if current_rev is None and not revisions:
+        return
+
+    current_reached_ref = [current_rev is None]
+
+    # Show migration history in chronological order
+    console.print("<base>")
+    for rev in revisions:
+        # Format and print the revision
+        console.print(format_revision(rev, current_rev, current_reached_ref))
+
+
 @db_cli.command()
 @click.option(
     "--message",