Class OracleSchemaEditor

tortoise/migrations/schema_editor/oracle.py:12–252

class OracleSchemaEditor(BaseSchemaEditor):
    DIALECT = "oracle"
    TABLE_CREATE_TEMPLATE = "CREATE TABLE {table_name} ({fields}){extra};"
    FIELD_TEMPLATE = '"{name}" {type} {nullable} {unique}{primary}'
    TABLE_COMMENT_TEMPLATE = "COMMENT ON TABLE {table} IS '{comment}';"
    COLUMN_COMMENT_TEMPLATE = "COMMENT ON COLUMN {table}.\"{column}\" IS '{comment}';"
    INDEX_CREATE_TEMPLATE = 'CREATE INDEX "{index_name}" ON {table_name} ({fields});'
    GENERATED_PK_TEMPLATE = '"{field_name}" {generated_sql}'
    FK_TEMPLATE = (
        '{constraint}FOREIGN KEY ("{db_column}")'
        ' REFERENCES {table} ("{field}") ON DELETE {on_delete}'
    )
    M2M_TABLE_TEMPLATE = (
        "CREATE TABLE {table_name} (\n"
        '    "{backward_key}" {backward_type} NOT NULL,\n'
        '    "{forward_key}" {forward_type} NOT NULL,\n'
        "    {backward_fk},\n"
        "    {forward_fk}\n"
        "){extra};"
    )
    DELETE_TABLE_TEMPLATE = "DROP TABLE {table} CASCADE CONSTRAINTS"
    DELETE_FIELD_TEMPLATE = 'ALTER TABLE {table} DROP COLUMN "{column}"'

    def __init__(self, connection, atomic: bool = True, collect_sql: bool = False) -> None:
        super().__init__(connection, atomic, collect_sql=collect_sql)
        self.comments_array: list[str] = []
        self._foreign_keys: list[str] = []

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

    def _get_table_comment_sql(self, table: str, comment: str) -> str:
        sql = self.TABLE_COMMENT_TEMPLATE.format(table=table, comment=self._escape_comment(comment))
        self.comments_array.append(sql)
        return ""

    def _get_column_comment_sql(self, table: str, column: str, comment: str) -> str:
        sql = self.COLUMN_COMMENT_TEMPLATE.format(
            table=table, column=column, comment=self._escape_comment(comment)
        )
        if sql not in self.comments_array:
            self.comments_array.append(sql)
        return ""

    def _post_table_hook(self) -> str:
        sql = "\n".join(self.comments_array)
        self.comments_array = []
        if sql:
            return "\n" + sql
        return ""

    def _get_fk_reference_string(
        self,
        constraint_name: str,
        db_field: str,