JezK

Edit File: oracle.py

from __future__ import annotations

from typing import Any
from typing import Optional
from typing import TYPE_CHECKING
from typing import Union

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql import sqltypes

from .base import AddColumn
from .base import alter_table
from .base import ColumnComment
from .base import ColumnDefault
from .base import ColumnName
from .base import ColumnNullable
from .base import ColumnType
from .base import format_column_name
from .base import format_server_default
from .base import format_table_name
from .base import format_type
from .base import IdentityColumnDefault
from .base import RenameTable
from .impl import DefaultImpl

if TYPE_CHECKING:
    from sqlalchemy.dialects.oracle.base import OracleDDLCompiler
    from sqlalchemy.engine.cursor import CursorResult
    from sqlalchemy.engine.cursor import LegacyCursorResult
    from sqlalchemy.sql.schema import Column

class OracleImpl(DefaultImpl):
    __dialect__ = "oracle"
    transactional_ddl = False
    batch_separator = "/"
    command_terminator = ""
    type_synonyms = DefaultImpl.type_synonyms + (
        {"VARCHAR", "VARCHAR2"},
        {"BIGINT", "INTEGER", "SMALLINT", "DECIMAL", "NUMERIC", "NUMBER"},
        {"DOUBLE", "FLOAT", "DOUBLE_PRECISION"},
    )
    identity_attrs_ignore = ()

def __init__(self, *arg, **kw) -> None:
        super(OracleImpl, self).__init__(*arg, **kw)
        self.batch_separator = self.context_opts.get(
            "oracle_batch_separator", self.batch_separator
        )

def _exec(
        self, construct: Any, *args, **kw
    ) -> Optional[Union["LegacyCursorResult", "CursorResult"]]:
        result = super(OracleImpl, self)._exec(construct, *args, **kw)
        if self.as_sql and self.batch_separator:
            self.static_output(self.batch_separator)
        return result

def emit_begin(self) -> None:
        self._exec("SET TRANSACTION READ WRITE")

def emit_commit(self) -> None:
        self._exec("COMMIT")

@compiles(AddColumn, "oracle")
def visit_add_column(
    element: "AddColumn", compiler: "OracleDDLCompiler", **kw
) -> str:
    return "%s %s" % (
        alter_table(compiler, element.table_name, element.schema),
        add_column(compiler, element.column, **kw),
    )

@compiles(ColumnNullable, "oracle")
def visit_column_nullable(
    element: "ColumnNullable", compiler: "OracleDDLCompiler", **kw
) -> str:
    return "%s %s %s" % (
        alter_table(compiler, element.table_name, element.schema),
        alter_column(compiler, element.column_name),
        "NULL" if element.nullable else "NOT NULL",
    )

@compiles(ColumnType, "oracle")
def visit_column_type(
    element: "ColumnType", compiler: "OracleDDLCompiler", **kw
) -> str:
    return "%s %s %s" % (
        alter_table(compiler, element.table_name, element.schema),
        alter_column(compiler, element.column_name),
        "%s" % format_type(compiler, element.type_),
    )

@compiles(ColumnName, "oracle")
def visit_column_name(
    element: "ColumnName", compiler: "OracleDDLCompiler", **kw
) -> str:
    return "%s RENAME COLUMN %s TO %s" % (
        alter_table(compiler, element.table_name, element.schema),
        format_column_name(compiler, element.column_name),
        format_column_name(compiler, element.newname),
    )

@compiles(ColumnDefault, "oracle")
def visit_column_default(
    element: "ColumnDefault", compiler: "OracleDDLCompiler", **kw
) -> str:
    return "%s %s %s" % (
        alter_table(compiler, element.table_name, element.schema),
        alter_column(compiler, element.column_name),
        "DEFAULT %s" % format_server_default(compiler, element.default)
        if element.default is not None
        else "DEFAULT NULL",
    )

@compiles(ColumnComment, "oracle")
def visit_column_comment(
    element: "ColumnComment", compiler: "OracleDDLCompiler", **kw
) -> str:
    ddl = "COMMENT ON COLUMN {table_name}.{column_name} IS {comment}"

comment = compiler.sql_compiler.render_literal_value(
        (element.comment if element.comment is not None else ""),
        sqltypes.String(),
    )

return ddl.format(
        table_name=element.table_name,
        column_name=element.column_name,
        comment=comment,
    )

@compiles(RenameTable, "oracle")
def visit_rename_table(
    element: "RenameTable", compiler: "OracleDDLCompiler", **kw
) -> str:
    return "%s RENAME TO %s" % (
        alter_table(compiler, element.table_name, element.schema),
        format_table_name(compiler, element.new_table_name, None),
    )

def alter_column(compiler: "OracleDDLCompiler", name: str) -> str:
    return "MODIFY %s" % format_column_name(compiler, name)

def add_column(compiler: "OracleDDLCompiler", column: "Column", **kw) -> str:
    return "ADD %s" % compiler.get_column_specification(column, **kw)

@compiles(IdentityColumnDefault, "oracle")
def visit_identity_column(
    element: "IdentityColumnDefault", compiler: "OracleDDLCompiler", **kw
):
    text = "%s %s " % (
        alter_table(compiler, element.table_name, element.schema),
        alter_column(compiler, element.column_name),
    )
    if element.default is None:
        # drop identity
        text += "DROP IDENTITY"
        return text
    else:
        text += compiler.visit_identity_column(element.default)
        return text

JezK

Edit File: oracle.py

Events