4f95ecd76bf8_normalize_distributed_query_architecture.py 文件源码

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    DistributedQueryTask = namedtuple('DistributedQueryTask', [
        'id', 'status', 'retrieved', 'guid', 'node_id'])

    distributed_query_task = op.create_table('distributed_query_task',
        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
        sa.Column('guid', sa.String(), nullable=False),
        sa.Column('status', sa.Integer(), nullable=False),
        sa.Column('timestamp', sa.DateTime(), nullable=True),
        sa.Column('distributed_query_id', sa.Integer(), nullable=False),
        sa.Column('node_id', sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(['distributed_query_id'], ['distributed_query.id'], ),
        sa.ForeignKeyConstraint(['node_id'], ['node.id'], ),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('guid')
    )

    cursor = op.get_bind().execute("""
        SELECT id, status, retrieved, guid, node_id
        FROM distributed_query
        ORDER BY id;"""
    )
    results = map(DistributedQueryTask._make, cursor.fetchall())
    distributed_query_tasks = [dict(
        distributed_query_id=r.id,
        status=r.status,
        timestamp=r.retrieved,
        guid=r.guid,
        node_id=r.node_id) for r in results]

    op.bulk_insert(distributed_query_task, distributed_query_tasks)

    op.add_column(u'distributed_query', sa.Column('description', sa.String(), nullable=True))
    op.drop_constraint(u'distributed_query_guid_key', 'distributed_query', type_='unique')
    op.drop_constraint(u'distributed_query_node_id_fkey', 'distributed_query', type_='foreignkey')
    op.drop_column(u'distributed_query', 'status')
    op.drop_column(u'distributed_query', 'retrieved')
    op.drop_column(u'distributed_query', 'guid')
    op.drop_column(u'distributed_query', 'node_id')
    op.add_column(u'distributed_query_result', sa.Column('distributed_query_task_id', sa.Integer(), nullable=True))

    # distributed queries and tasks were the same before,
    # so their id's will remain the same as well.
    op.execute("""
        UPDATE distributed_query_result
        SET distributed_query_task_id = distributed_query_id;"""
    )

    op.alter_column(u'distributed_query_result', 'distributed_query_task_id', nullable=False)
    op.create_foreign_key(None, 'distributed_query_result', 'distributed_query_task', ['distributed_query_task_id'], ['id'])
    ### end Alembic commands ###