Only treat a missing migrations table as no applied migrations

CHANGELOG.rst

@@ -16,6 +16,10 @@ Added
 - Support forward relation ``in`` and ``not_in`` filters with model instances. (#2323)
 - SQLite now persists ``TimeField`` values as ISO-8601 times. (#2297)
 
+Changed
+^^^^^
+- ``MigrationRecorder.applied_migrations()`` now checks for the migrations table before querying history, allowing catalog lookup errors to propagate while retaining the empty-list fallback for history query errors. (#2382)
+
 Fixed
 ^^^^^
 - ``DecimalField`` no longer raises ``decimal.InvalidOperation`` for values with more than 28 significant digits. (#2349)

tests/migrations/test_recorder.py

@@ -3,12 +3,25 @@
 import pytest
 
 from tortoise.backends.base.client import Capabilities
+from tortoise.backends.mysql import MySQLClient
+from tortoise.exceptions import DBConnectionError, OperationalError
+from tortoise.migrations.executor import MigrationExecutor
 from tortoise.migrations.recorder import MigrationRecorder
 
+RECORDER_DIALECTS = ["sqlite", "postgres", "mysql", "mssql", "oracle"]
+
 
 class FakeConnection:
-    def __init__(self, dialect: str) -> None:
+    def __init__(
+        self,
+        dialect: str,
+        *,
+        table_exists: bool = True,
+        error: Exception | None = None,
+    ) -> None:
         self.capabilities = Capabilities(dialect)
+        self.table_exists = table_exists
+        self.error = error
         self.executed_scripts: list[str] = []
         self.inserts: list[tuple[str, list]] = []
         self.queries: list[tuple[str, list | None]] = []
@@ -22,6 +35,10 @@ async def execute_insert(self, query: str, values: list) -> int:
 
     async def execute_query(self, query: str, values: list | None = None):
         self.queries.append((query, values))
+        if self.error is not None:
+            raise self.error
+        if query.startswith("SELECT 1 "):  # catalog lookup for the migrations table
+            return None, [{"1": 1}] if self.table_exists else []
         return None, []
 
 
@@ -42,7 +59,7 @@ async def test_recorder_quotes_mysql_identifiers() -> None:
     assert insert_values[1] == "0001_initial"
 
     assert connection.queries
-    select_query = connection.queries[0][0]
+    select_query = connection.queries[-1][0]
     assert "SELECT `app`, `name`" in select_query
 
 
@@ -159,3 +176,59 @@ async def test_recorder_oracle_placeholders() -> None:
     delete_query = connection.queries[0][0]
     assert '"app" = ?' in delete_query
     assert '"name" = ?' in delete_query
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("dialect", RECORDER_DIALECTS)
+async def test_recorder_missing_table_means_no_migrations(dialect: str) -> None:
+    connection = FakeConnection(dialect, table_exists=False)
+    recorder = MigrationRecorder(connection)
+
+    assert await recorder.applied_migrations() == []
+    # Only the catalog lookup ran; the migrations table itself was never queried.
+    assert len(connection.queries) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("dialect", RECORDER_DIALECTS)
+async def test_recorder_propagates_connection_errors(dialect: str) -> None:
+    error = DBConnectionError("sorry, too many clients already")
+    recorder = MigrationRecorder(FakeConnection(dialect, error=error))
+
+    with pytest.raises(DBConnectionError):
+        await recorder.applied_migrations()
+
+
+@pytest.mark.asyncio
+async def test_recorder_query_errors_mean_no_migrations_when_table_exists() -> None:
+    class DeniedSelectConnection(FakeConnection):
+        async def execute_query(self, query: str, values: list | None = None):
+            if not query.startswith("SELECT 1 "):
+                raise OperationalError("permission denied for table tortoise_migrations")
+            return await super().execute_query(query, values)
+
+    recorder = MigrationRecorder(DeniedSelectConnection("postgres"))
+
+    assert await recorder.applied_migrations() == []
+
+
+@pytest.mark.asyncio
+async def test_recorder_detects_migrations_table_on_real_database(db_isolated) -> None:
+    connection = db_isolated.db()
+    if (
+        isinstance(connection, MySQLClient)
+        and connection.storage_engine
+        and connection.storage_engine.lower() != "innodb"
+    ):
+        pytest.skip(
+            "The migrations table's unique key exceeds MyISAM's 1000-byte key limit, "
+            "a pre-existing limitation unrelated to this test"
+        )
+    executor = MigrationExecutor(connection, {})
+    recorder = executor.recorder
+
+    assert await recorder.applied_migrations() == []
+
+    # migrate() reads the history before and after creating the migrations table.
+    await executor.migrate()
+    assert await recorder.applied_migrations() == []

tortoise/migrations/recorder.py

@@ -69,12 +69,51 @@ async def ensure_schema(self, schema_editor) -> None:
                 return
             raise
 
+    async def _table_exists(self) -> bool:
+        """Report whether the migrations table exists, using the dialect's catalog.
+
+        A missing table means no migrations have been applied yet. Errors from this
+        catalog lookup propagate instead of being mistaken for an empty migration history.
+        """
+        if self._dialect == "sqlite":
+            query = "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?"
+            values: list | None = [self.table_name]
+        elif self._dialect == "postgres":
+            # to_regclass resolves the name through search_path, like the unqualified SELECT.
+            # The name is inlined because asyncpg and psycopg use different placeholders.
+            regclass = self._quote(self.table_name).replace("'", "''")
+            query = f"SELECT 1 WHERE to_regclass('{regclass}') IS NOT NULL"  # nosec B608
+            values = None
+        elif self._dialect == "mysql":
+            query = (
+                "SELECT 1 FROM information_schema.tables "
+                "WHERE table_schema = DATABASE() AND table_name = %s"
+            )
+            values = [self.table_name]
+        elif self._dialect == "mssql":
+            # OBJECT_ID resolves the name like the unqualified SELECT (default schema, then dbo).
+            query = "SELECT 1 WHERE OBJECT_ID(?, 'U') IS NOT NULL"
+            values = [self._quote(self.table_name)]
+        elif self._dialect == "oracle":
+            query = (
+                "SELECT 1 FROM all_tables "
+                "WHERE owner = SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA') AND table_name = ?"
+            )
+            values = [self.table_name]
+        else:
+            # Unknown dialect: fall back to querying the migrations table.
+            return True
+        _, rows = await self.connection.execute_query(query, values)
+        return bool(rows)
+
     async def applied_migrations(self) -> list[MigrationKey]:
         query = (
             f"SELECT {self._quote('app')}, {self._quote('name')} "  # nosec B608
             f"FROM {self._quote(self.table_name)} "
             f"ORDER BY {self._quote('applied_at')}, {self._quote('app')}, {self._quote('name')}"
         )
+        if not await self._table_exists():
+            return []
         try:
             _, rows = await self.connection.execute_query(query)
         except Exception: