Python alembic.op 模块，batch_alter_table() 实例源码

def upgrade():
    table_prefix = context.config.get_main_option('table_prefix')
    table_name = table_prefix + 'environment_hierarchy_level_value'
    with op.batch_alter_table(table_name) as batch:
        batch.drop_column('parent_id')

        batch.drop_constraint(
            table_name + '_level_id_fkey',
            type_='foreignkey'
        )
        batch.create_foreign_key(
            table_name + '_level_id_fkey',
            table_prefix + 'environment_hierarchy_level',
            ['level_id'], ['id'], ondelete='CASCADE'
        )

        batch.create_unique_constraint(
            table_name + '_level_id_value_unique',
            ['level_id', 'value']
        )

def upgrade():
    table_prefix = context.config.get_main_option('table_prefix')
    op.drop_table(table_prefix + 'template')
    table_name = table_prefix + 'environment_schema_values'
    with op.batch_alter_table(table_name) as batch:
        batch.drop_constraint(table_name + '_schema_id_fkey', 'foreignkey')
        batch.alter_column(
            'schema_id',
            new_column_name='resource_definition_id',
            existing_type=sa.Integer(),
        )
    op.rename_table(table_name, table_prefix + 'resource_values')
    op.rename_table(table_prefix + 'schema',
                    table_prefix + 'resource_definition')
    with op.batch_alter_table(table_prefix + 'resource_definition') as batch:
        batch.drop_column('namespace_id')
    op.drop_table(table_prefix + 'namespace')
    table_name = table_prefix + 'resource_values'
    with op.batch_alter_table(table_name) as batch:
        batch.create_foreign_key(
            table_name + '_resource_definition_id_fkey',
            table_prefix + 'resource_definition',
            ['resource_definition_id'],
            ['id'],
        )

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('schedule_info',
    sa.Column('id', sa.Integer(), autoincrement=False, nullable=False),
    sa.Column('last_changed_at', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('schedule_meta',
    sa.Column('parent_id', sa.Integer(), nullable=False),
    sa.Column('last_run_at', sa.DateTime(), nullable=True),
    sa.Column('total_run_count', sa.Integer(), nullable=True),
    sa.ForeignKeyConstraint(['parent_id'], ['schedule_task.id'], ),
    sa.PrimaryKeyConstraint('parent_id')
    )
    with op.batch_alter_table('schedule_task') as batch_op:
        batch_op.drop_column('total_run_count')
        batch_op.drop_column('last_run_at')
    # ### end Alembic commands ###

def create_foreign_key(
            self, name, referent, local_cols, remote_cols, **kw):
        """Issue a "create foreign key" instruction using the
        current batch migration context.

        The batch form of this call omits the ``source`` and ``source_schema``
        arguments from the call.

        e.g.::

            with batch_alter_table("address") as batch_op:
                batch_op.create_foreign_key(
                            "fk_user_address",
                            "user", ["user_id"], ["id"])

        .. seealso::

            :meth:`.Operations.create_foreign_key`

        """
        return super(BatchOperations, self).create_foreign_key(
            name, self.impl.table_name, referent, local_cols, remote_cols,
            source_schema=self.impl.schema, **kw)

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    op.drop_column('users', 'last_name')
    op.drop_column('users', 'first_name')
    ### end Alembic commands ###
    with op.batch_alter_table("notes") as batch_op:
        batch_op.drop_constraint(
            "notes_author_id_fkey", type_="foreignkey")
    with op.batch_alter_table("notebooks") as batch_op:
        batch_op.drop_constraint(
            "notebooks_author_id_fkey", type_="foreignkey")
    op.create_foreign_key(
        "notes_author_id_fkey", "notes", "users", 
        ["author_id"], ["id"], ondelete="CASCADE")
    op.create_foreign_key(
        "notebooks_author_id_fkey", "notebooks", "users", 
        ["author_id"], ["id"], ondelete="CASCADE")

def upgrade():
    table_prefix = context.config.get_main_option('table_prefix')
    table_name = table_prefix + 'component'
    with op.batch_alter_table(table_name) as batch:
        batch.create_unique_constraint(
            table_name + '_component_name_unique',
            ['name'],
        )

def downgrade():
    table_prefix = context.config.get_main_option('table_prefix')
    table_name = table_prefix + 'component'
    with op.batch_alter_table(table_name) as batch:
        batch.drop_constraint(
            table_name + '_component_name_unique',
            type_='unique',
        )

def upgrade():
    table_prefix = context.config.get_main_option('table_prefix')
    table_name = table_prefix + 'resource_values'
    with op.batch_alter_table(table_name) as batch:
        batch.alter_column(
            'values',
            server_default='{}',
            existing_type=tuning_box.db.Json(),
        )

def downgrade():
    table_prefix = context.config.get_main_option('table_prefix')
    table_name = table_prefix + 'resource_values'
    with op.batch_alter_table(table_name) as batch:
        batch.alter_column(
            'values',
            server_default=None,
            existing_type=tuning_box.db.Json(),
        )

def upgrade():
    table_prefix = context.config.get_main_option('table_prefix')
    table_name = table_prefix + 'environment_hierarchy_level'

    with op.batch_alter_table(table_name) as batch:
        batch.drop_constraint(
            table_prefix + 'environment_hierarchy_level_environment_id_fkey',
            type_='foreignkey'
        )
        batch.create_foreign_key(
            table_prefix + 'environment_hierarchy_level_environment_id_fkey',
            table_prefix + 'environment',
            ['environment_id'], ['id'], ondelete='CASCADE'
        )

def downgrade():
    table_prefix = context.config.get_main_option('table_prefix')
    table_name = table_prefix + 'environment_hierarchy_level'
    with op.batch_alter_table(table_name) as batch:
        batch.drop_constraint(
            table_prefix + 'environment_hierarchy_level_environment_id_fkey',
            type_='foreignkey'
        )
        batch.create_foreign_key(
            table_prefix + 'environment_hierarchy_level_environment_id_fkey',
            table_prefix + 'environment',
            ['environment_id'], ['id']
        )

def upgrade():
    with op.batch_alter_table('pages') as batch_op:
        batch_op.drop_column('is_html')
        batch_op.drop_column('use_layout_header')

def downgrade():
    with op.batch_alter_table('pages') as batch_op:
            batch_op.add_column(
                sa.Column('is_html', sa.Boolean(), nullable=True)
            )
            batch_op.add_column(
                sa.Column('use_layout_header', sa.Boolean(), nullable=True)
            )

def upgrade():
    """Upgrade database."""
    # The following is a ridiculous hack to force table recreation for SQLite to
    # enable the use of a default timestamp.
    recreate = 'auto'
    migrate_context = context.get_context()
    sqlite_dialect_class = None
    if getattr(sa.dialects, 'sqlite', False):
        sqlite_dialect_class = (sa.dialects.sqlite.pysqlite
                                .SQLiteDialect_pysqlite)
    if migrate_context.dialect.__class__ == sqlite_dialect_class:
        recreate = 'always'
    with op.batch_alter_table('games', recreate=recreate) as batch_op:
        batch_op.add_column(sa.Column('played_at', sa.DateTime(),
                            nullable=False, server_default=sa.func.now()))

def downgrade():
    """Downgrade database."""
    with op.batch_alter_table('games') as batch_op:
        batch_op.drop_column('played_at')

def upgrade():
    """Upgrade database."""
    # The following is a ridiculous hack to force table recreation for SQLite to
    # enable the use of a default timestamp.
    recreate = 'auto'
    migrate_context = context.get_context()
    sqlite_dialect_class = None
    if getattr(sa.dialects, 'sqlite', False):
        sqlite_dialect_class = (sa.dialects.sqlite.pysqlite
                                .SQLiteDialect_pysqlite)
    if migrate_context.dialect.__class__ == sqlite_dialect_class:
        recreate = 'always'
    with op.batch_alter_table('games', recreate=recreate) as batch_op:
        batch_op.add_column(sa.Column('last_modified_at', sa.DateTime(),
                            nullable=False, server_default=sa.func.now()))

def downgrade():
    """Downgrade database."""
    with op.batch_alter_table('games') as batch_op:
        batch_op.drop_column('last_modified_at')

def upgrade():
    """Upgrade database."""
    # The following is a ridiculous hack to force table recreation for SQLite to
    # enable the use of a default timestamp.
    recreate = 'auto'
    migrate_context = context.get_context()
    sqlite_dialect_class = None
    if getattr(sa.dialects, 'sqlite', False):
        sqlite_dialect_class = (sa.dialects.sqlite.pysqlite
                                .SQLiteDialect_pysqlite)
    if migrate_context.dialect.__class__ == sqlite_dialect_class:
        recreate = 'always'
    with op.batch_alter_table('games', recreate=recreate) as batch_op:
        batch_op.add_column(sa.Column('created_at', sa.DateTime(),
                            nullable=False, server_default=sa.func.now()))

def upgrade():
    with op.batch_alter_table('container', schema=None) as batch_op:
        batch_op.add_column(sa.Column('interactive', sa.Boolean(),
                                      nullable=True))
        batch_op.drop_column('tty')
        batch_op.drop_column('stdin_open')

def upgrade():
    # batch_alter_table() is necessary here because SQLite does not support
    # the SQL ALTER statement which would normally be emitted here.

    # What this function does is basically cause all the data in the old
    # version of the table to be copied into a new table, then delete the old
    # table at the end if everything works as planned.
    with op.batch_alter_table('menu_entries') as batch_op:
        batch_op.create_unique_constraint(
            constraint_name="unique_menu_entry_date_mensa_category_description",
            columns=['date_valid', 'mensa', 'category', 'description'])

def downgrade():
    with op.batch_alter_table('menu_entries') as batch_op:
        batch_op.drop_constraint(
            constraint_name="unique_menu_entry_date_mensa_category_description",
            type_='unique')

def upgrade():
    # There can be data truncation here as LargeBinary can be smaller than the pickle 
    # type.

    # use batch_alter_table to support SQLite workaround
    with op.batch_alter_table("xcom") as batch_op:
        batch_op.alter_column('value', type_=sa.LargeBinary())

def downgrade():
    # use batch_alter_table to support SQLite workaround
    with op.batch_alter_table("xcom") as batch_op:
        batch_op.alter_column('value', type_=sa.PickleType(pickler=dill))

def upgrade():
    # use batch_alter_table to support SQLite workaround
    with op.batch_alter_table("task_instance") as batch_op:
        batch_op.alter_column('duration',
                              existing_type=mysql.INTEGER(display_width=11),
                              type_=sa.Float(),
                              existing_nullable=True)

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table('users', schema=None) as batch_op:
        batch_op.add_column(sa.Column('ldap_auth', sa.Boolean(), server_default=sa.text('false'), nullable=False))

    ### end Alembic commands ###

def downgrade():
    ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table('users', schema=None) as batch_op:
        batch_op.drop_column('ldap_auth')

    ### end Alembic commands ###

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table('menus', schema=None) as batch_op:
        batch_op.drop_column('location')

    ### end Alembic commands ###

def downgrade():
    ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table('menus', schema=None) as batch_op:
        batch_op.add_column(sa.Column('location', sa.VARCHAR(length=256), nullable=False))

    ### end Alembic commands ###

def upgrade():
    with op.batch_alter_table(t1_name) as batch_op:
        batch_op.add_column(sa.Column('enabled',
                                      sa.Boolean(),
                                      nullable=False,
                                      server_default='1'))

    with op.batch_alter_table(t2_name) as batch_op:
        batch_op.add_column(sa.Column('do_not_use',
                                      sa.Boolean(),
                                      nullable=False,
                                      server_default='0'))

def downgrade():
    """alexm: i believe this method is never called"""
    with op.batch_alter_table(t2_name) as batch_op:
        batch_op.drop_column('do_not_use')

    with op.batch_alter_table(t1_name) as batch_op:
        batch_op.drop_column('enabled')

def upgrade():
    with op.batch_alter_table('editions', schema=None) as batch_op:
        batch_op.add_column(sa.Column('mode', sa.Integer(), nullable=True))

def downgrade():
    with op.batch_alter_table('editions', schema=None) as batch_op:
        batch_op.drop_column('mode')

def upgrade():
    with op.batch_alter_table('products', schema=None) as batch_op:
        batch_op.add_column(sa.Column('surrogate_key',
                                      sa.String(length=32),
                                      nullable=True))

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table('restaurants_menuitem', schema=None) as batch_op:
        batch_op.alter_column('price',
               existing_type=sa.NUMERIC(precision=2, scale=0),
               type_=sa.Integer(),
               existing_nullable=False)

    ### end Alembic commands ###

def downgrade():
    ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table('restaurants_menuitem', schema=None) as batch_op:
        batch_op.alter_column('price',
               existing_type=sa.Integer(),
               type_=sa.NUMERIC(precision=2, scale=0),
               existing_nullable=False)

    ### end Alembic commands ###

def upgrade():
    # use batch_alter_table to support SQLite workaround
    with op.batch_alter_table("task_instance") as batch_op:
        batch_op.alter_column('duration',
                              existing_type=mysql.INTEGER(display_width=11),
                              type_=sa.Float(),
                              existing_nullable=True)

def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table('user') as batch_op:
        batch_op.drop_column('state')
        batch_op.drop_column('last_active_at')
        batch_op.drop_column('is_active')
        # ### end Alembic commands ###

def upgrade():
    with op.batch_alter_table('fda_dap') as batch_op:
        batch_op.add_column(sa.Column('drug_name', sa.Text))
        batch_op.add_column(sa.Column('active_ingredients', sa.Text))
        batch_op.add_column(sa.Column('company', sa.Text))

def downgrade():
    with op.batch_alter_table('fda_dap') as batch_op:
        batch_op.drop_column('drug_name')
        batch_op.drop_column('active_ingredients')
        batch_op.drop_column('company')

def upgrade():
    with op.batch_alter_table('upgrade_tasks') as batch_op:
        batch_op.add_column(
            sa.Column('status', sa.Enum(*constants.UPGRADE_STATUSES)))

def downgrade():
    with op.batch_alter_table('upgrade_tasks') as batch_op:
        batch_op.drop_column('status')

def upgrade():
    op.create_table(
        'hosts_services',
        sa.Column('host_id', sa.String(36), sa.ForeignKey('hosts.id')),
        sa.Column('service_id', sa.String(36), sa.ForeignKey('services.id')),
        sa.UniqueConstraint('host_id', 'service_id'))

    with op.batch_alter_table('services') as batch_op:
        batch_op.drop_constraint('fk_services_host_id_hosts')
        batch_op.drop_column('host_id')

def upgrade():
    connection = op.get_bind()
    # spam_threshold is a X/15 based value, we're converting it to percent.
    for user in connection.execute(user_table.select()):
         connection.execute(
            user_table.update().where(
                user_table.c.email == user.email
            ).values(
                spam_threshold=int(100. * float(user.spam_threshold or 0.) / 15.)
            )
         )
    # set default to 80%
    with op.batch_alter_table('user') as batch:
        batch.alter_column('spam_threshold', default=80.)

def downgrade():
    connection = op.get_bind()
    # spam_threshold is a X/15 based value, we're converting it from percent.
    for user in connection.execute(user_table.select()):
         connection.execute(
            user_table.update().where(
                user_table.c.email == user.email
            ).values(
                spam_threshold=int(15. * float(user.spam_threshold or 0.) / 100.)
            )
         )
    # set default to 10/15
    with op.batch_alter_table('user') as batch:
        batch.alter_column('spam_threshold', default=10.)

def downgrade():
    with op.batch_alter_table('domain') as batch:
        batch.drop_column('max_quota_bytes')

def upgrade():
    with op.batch_alter_table('user') as batch:
        batch.add_column(sa.Column('forward_keep', sa.Boolean(), nullable=False, server_default=sa.sql.expression.true()))

def downgrade():
    with op.batch_alter_table('user') as batch:
        batch.drop_column('forward_keep')

def upgrade():
    with op.batch_alter_table('user') as batch:
        batch.alter_column('email', type_=sa.String(length=255, collation="NOCASE"))
    with op.batch_alter_table('alias') as batch:
        batch.alter_column('email', type_=sa.String(length=255, collation="NOCASE"))

def downgrade():
    with op.batch_alter_table('user') as batch:
        batch.alter_column('email', type_=sa.String(length=255))
    with op.batch_alter_table('alias') as batch:
        batch.alter_column('email', type_=sa.String(length=255))

def downgrade():
    with op.batch_alter_table('fetch') as batch:
        batch.drop_column('keep')