Method collect_sql

tortoise/migrations/executor.py:156–194

Collect SQL statements for a single migration without executing them. Args: app_label: The application label. migration_name: The migration name (exact or prefix match). backward: If True, collect SQL for unapplying the migration. Returns:

(
        self,
        app_label: str,
        migration_name: str,
        *,
        backward: bool = False,
    )

        return BaseSchemaEditor(self.connection, atomic=atomic, collect_sql=collect_sql)

    async def collect_sql(
        self,
        app_label: str,
        migration_name: str,
        *,
        backward: bool = False,
    ) -> list[str]:
        """Collect SQL statements for a single migration without executing them.

        Args:
            app_label: The application label.
            migration_name: The migration name (exact or prefix match).
            backward: If True, collect SQL for unapplying the migration.

        Returns:
            A list of SQL strings (including comment annotations).
        """
        await self.loader.build_graph()

        # Resolve migration key — support prefix matching
        key = self._resolve_migration_key(app_label, migration_name)
        migration = self.loader.graph.nodes[key]
        if not isinstance(migration, Migration):
            raise ValueError(f"Missing migration for {key}")

        # For SQL collection we don't need the real database — treat all
        # migrations in the graph as "applied" so _project_state replays
        # them up to the target.
        all_keys = set(self.loader.graph.nodes.keys())
        state = await self._project_state(all_keys, upto=key)

        editor = self._schema_editor(atomic=False, collect_sql=True)

        if backward:
            await migration.unapply(state, schema_editor=editor, collect_sql=True)
        else:
            await migration.apply(state, schema_editor=editor, collect_sql=True)

        return editor.collected_sql

    def _resolve_migration_key(self, app_label: str, migration_name: str) -> MigrationKey:
        """Resolve a migration name to a MigrationKey, supporting prefix matching."""