#NEWLINE# def _change_deleted_column_type_to_id_type_sqlite( migrate_engine , table_name , **col_name_col_instance ) : #NEWLINE# #INDENT# insp = reflection . Inspector . from_engine ( migrate_engine ) #NEWLINE# meta = MetaData ( bind = migrate_engine ) #NEWLINE# table = Table ( table_name , meta , autoload = True ) #NEWLINE# default_deleted_value = _get_default_deleted_value ( table ) #NEWLINE# columns = [ ] #NEWLINE# for column in table . columns : #NEWLINE# #INDENT# column_copy = None #NEWLINE# if ( column . name != 'deleted' ) : #NEWLINE# #INDENT# if isinstance ( column . type , NullType ) : #NEWLINE# #INDENT# column_copy = _get_not_supported_column ( col_name_col_instance , column . name ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# column_copy = column . copy ( ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# column_copy = Column ( 'deleted' , table . c . id . type , default = default_deleted_value ) #NEWLINE# #UNINDENT# columns . append ( column_copy ) #NEWLINE# #UNINDENT# def is_deleted_column_constraint( constraint ) : #NEWLINE# #INDENT# if ( not isinstance ( constraint , CheckConstraint ) ) : #NEWLINE# #INDENT# return False #NEWLINE# #UNINDENT# sqltext = str ( constraint . sqltext ) #NEWLINE# constraint_markers = ( 'deleted in (0, 1)' , 'deleted IN (:deleted_1, :deleted_2)' , 'deleted IN (:param_1, :param_2)' ) #NEWLINE# return any ( ( sqltext . endswith ( marker ) for marker in constraint_markers ) ) #NEWLINE# #UNINDENT# constraints = [ ] #NEWLINE# for constraint in table . constraints : #NEWLINE# #INDENT# if ( not is_deleted_column_constraint ( constraint ) ) : #NEWLINE# #INDENT# constraints . append ( constraint . copy ( ) ) #NEWLINE# #UNINDENT# #UNINDENT# new_table = Table ( ( table_name + '__tmp__' ) , meta , * ( columns + constraints ) ) #NEWLINE# new_table . create ( ) #NEWLINE# indexes = [ ] #NEWLINE# for index in insp . get_indexes ( table_name ) : #NEWLINE# #INDENT# column_names = [ new_table . c [ c ] for c in index [ 'column_names' ] ] #NEWLINE# indexes . append ( Index ( index [ 'name' ] , * column_names , unique = index [ 'unique' ] ) ) #NEWLINE# #UNINDENT# ins = InsertFromSelect ( is_deleted_column_constraint , table . select ( ) ) #NEWLINE# migrate_engine . execute ( ins ) #NEWLINE# table . drop ( ) #NEWLINE# [ index . create ( migrate_engine ) for index in indexes ] #NEWLINE# new_table . rename ( table_name ) #NEWLINE# new_table . update ( ) . where ( new_table . c . deleted ) . values ( deleted = new_table . c . id ) . execute ( ) #NEWLINE# new_table . update ( ) . where ( ( not new_table . c . deleted ) ) . values ( deleted = default_deleted_value ) . execute ( )