171 lines
5.2 KiB
Python
171 lines
5.2 KiB
Python
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
|