Class SqliteSchemaEditor

tortoise/migrations/schema_editor/sqlite.py:18–427

class SqliteSchemaEditor(SqliteQuotingMixin, BaseSchemaEditor):
    DIALECT = "sqlite"
    DELETE_TABLE_TEMPLATE = "DROP TABLE {table}"
    DELETE_FIELD_TEMPLATE = 'ALTER TABLE {table} DROP COLUMN "{column}"'
    DROP_INDEX_TEMPLATE = 'DROP INDEX "{name}"'
    RENAME_INDEX_TEMPLATE = None

    @classmethod
    def _get_escape_translation_table(cls) -> list[str]:
        table = super()._get_escape_translation_table()
        table[ord('"')] = '"'
        table[ord("'")] = "'"
        table[ord("/")] = "\\/"
        return table

    async def _run_sql(self, sql: str) -> None:
        """Execute DDL SQL on SQLite.

        In atomic mode, uses execute_query() per statement because
        sqlite3.executescript() issues an implicit COMMIT.

        In collect_sql mode, delegates to the base class to collect
        the SQL without splitting or executing.
        """
        if self.collect_sql:
            await super()._run_sql(sql)
            return
        if self.atomic_migration:
            for statement in sql.split(";"):
                statement = statement.strip()
                if statement:
                    await self.client.execute_query(statement)
        else:
            await self.client.execute_script(sql)

    def _get_table_comment_sql(self, table: str, comment: str) -> str:
        return f" /* {self._escape_comment(comment)} */"

    def _get_column_comment_sql(self, table: str, column: str, comment: str) -> str:
        return f" /* {self._escape_comment(comment)} */"

    async def add_field(self, model, field_name: str) -> None:
        field = model._meta.fields_map[field_name]
        if isinstance(field, ManyToManyFieldInstance):
            table_string = self._get_m2m_table_definition(model, field)
            if table_string:
                await self._run_sql(table_string)
            return
        qualified_table = self._qualify_table_name(model._meta.db_table, model._meta.schema)
        if isinstance(field, ForeignKeyFieldInstance):
            key_field_name = field.source_field or field_name
            db_field = model._meta.fields_db_projection.get(key_field_name, key_field_name)
            key_field = model._meta.fields_map[key_field_name]
            fk_field = cast(ForeignKeyFieldInstance, key_field.reference)
            comment = (
                self._get_column_comment_sql(
                    table=qualified_table,
                    column=db_field,