Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return cls(tname, col.name, _orig_column=col)
def to_column(self):
if self._orig_column is not None:
return self._orig_column
return Column(self.column_name, **self.kw)
@classmethod
def drop_column(cls, table_name, column_name, **kw):
return cls(table_name, column_name, **kw)
def run(self):
self.engine.drop_column(self.table_name, self.column_name)
@Migration.register_operation("alter_column")
class AlterColumnOp(AlterTableOp):
def __init__(
self, table_name, column_name,
existing_type=None,
existing_default=None,
existing_notnull=None,
modify_notnull=None,
modify_default=DEFAULT_VALUE,
modify_name=None,
modify_type=None,
**kw
):
super(AlterColumnOp, self).__init__(table_name)
self.column_name = column_name
self.existing_type = existing_type
**self.table_kw)
@classmethod
def drop_table(cls, table_name, **kw):
return cls(table_name, table_kw=kw)
def run(self):
self.engine.drop_table(self.table_name)
class AlterTableOp(Operation):
def __init__(self, table_name):
self.table_name = table_name
@Migration.register_operation("rename_table")
class RenameTableOp(AlterTableOp):
def __init__(self, old_table_name, new_table_name):
super(RenameTableOp, self).__init__(old_table_name)
self.new_table_name = new_table_name
@classmethod
def rename_table(cls, old_table_name, new_table_name):
return cls(old_table_name, new_table_name)
def run(self):
raise NotImplementedError(
'Table renaming is currently not supported.'
)
@Migration.register_operation("add_column")
class RenameTableOp(AlterTableOp):
def __init__(self, old_table_name, new_table_name):
super(RenameTableOp, self).__init__(old_table_name)
self.new_table_name = new_table_name
@classmethod
def rename_table(cls, old_table_name, new_table_name):
return cls(old_table_name, new_table_name)
def run(self):
raise NotImplementedError(
'Table renaming is currently not supported.'
)
@Migration.register_operation("add_column")
class AddColumnOp(AlterTableOp):
def __init__(self, table_name, column):
super(AddColumnOp, self).__init__(table_name)
self.column = column
def reverse(self):
return DropColumnOp.from_column_and_tablename(
self.table_name, self.column)
def to_diff_tuple(self):
return ("add_column", self.table_name, self.column)
def to_column(self):
return self.column
@classmethod
))
)
class MigrationOp(Operation):
def __init__(self, rev_id, upgrade_ops, downgrade_ops, message=None,
head=None, splice=None):
self.rev_id = rev_id
self.message = message
self.head = head
self.splice = splice
self.upgrade_ops = upgrade_ops
self.downgrade_ops = downgrade_ops
@Migration.register_operation("create_table")
class CreateTableOp(Operation):
def __init__(self, table_name, columns, _orig_table=None, **kw):
self.table_name = table_name
self.columns = columns
self.kw = kw
self._orig_table = _orig_table
def reverse(self):
return DropTableOp.from_table(self.to_table())
def to_diff_tuple(self):
return ("add_table", self.to_table())
@classmethod
def from_table(cls, table):
return cls(
self.unique, **self.kw)
@classmethod
def create_index(
cls, index_name, table_name, fields=[], expressions=[], unique=False,
**kw
):
return cls(index_name, table_name, fields, expressions, unique, **kw)
def run(self):
self.engine.create_index(
self.index_name, self.table_name, self.fields, self.expressions,
self.unique, **self.kw)
@Migration.register_operation("drop_index")
class DropIndexOp(Operation):
def __init__(self, index_name, table_name=None, _orig_index=None):
self.index_name = index_name
self.table_name = table_name
self._orig_index = _orig_index
def to_diff_tuple(self):
return ("remove_index", self.to_index())
def reverse(self):
if self._orig_index is None:
raise ValueError(
"operation is not reversible; "
"original index is not present")
return CreateIndexOp.from_index(self._orig_index)
def to_column(self):
return self.column
@classmethod
def from_column_and_tablename(cls, tname, col):
return cls(tname, col)
@classmethod
def add_column(cls, table_name, column):
return cls(table_name, column)
def run(self):
self.engine.add_column(self.table_name, self.column)
@Migration.register_operation("drop_column")
class DropColumnOp(AlterTableOp):
def __init__(self, table_name, column_name, _orig_column=None, **kw):
super(DropColumnOp, self).__init__(table_name)
self.column_name = column_name
self.kw = kw
self._orig_column = _orig_column
def to_diff_tuple(self):
return ("remove_column", self.table_name, self.to_column())
def reverse(self):
if self._orig_column is None:
raise ValueError(
"operation is not reversible; "
"original column is not present")
if self._orig_table is not None:
return self._orig_table
from .generation import MetaTable
return MetaTable(
self.table_name, self.columns, **self.kw
)
@classmethod
def create_table(cls, table_name, *columns, **kw):
return cls(table_name, columns, **kw)
def run(self):
self.engine.create_table(self.table_name, self.columns, **self.kw)
@Migration.register_operation("drop_table")
class DropTableOp(Operation):
def __init__(self, table_name, table_kw=None, _orig_table=None):
self.table_name = table_name
self.table_kw = table_kw or {}
self._orig_table = _orig_table
def to_diff_tuple(self):
return ("remove_table", self.to_table())
def reverse(self):
if self._orig_table is None:
raise ValueError(
"operation is not reversible; "
"original table is not present")
return CreateTableOp.from_table(self._orig_table)
existing_type=existing_type,
existing_default=existing_default,
existing_notnull=existing_notnull,
modify_name=new_column_name,
modify_type=type,
modify_default=default,
modify_notnull=notnull,
**kw
)
def run(self):
self.engine.alter_column(
self.table_name, self.column_name, self.to_diff_tuple())
@Migration.register_operation("create_index")
class CreateIndexOp(Operation):
def __init__(
self, index_name, table_name, fields=[], expressions=[], unique=False,
_orig_index=None, **kw
):
self.index_name = index_name
self.table_name = table_name
self.fields = fields
self.expressions = expressions
self.unique = unique
self.kw = kw
self._orig_index = _orig_index
def reverse(self):
return DropIndexOp.from_index(self.to_index())