4f95ecd76bf8_normalize_distributed_query_architecture.py 文件源码

def downgrade():
    ### commands auto generated by Alembic - please adjust! ###
    DistributedQuery = namedtuple('DistributedQuery', [
        'task_id', 'query_id',
        'guid', 'status', 'sql', 'timestamp', 'not_before',
        'retrieved', 'node_id'])

    cursor = op.get_bind().execute("""
        SELECT DISTINCT t.id AS task_id, q.id AS query_id,
            t.guid, t.status, q.sql, q.timestamp, q.not_before,
            t.timestamp AS retrieved, t.node_id
        FROM distributed_query q
        INNER JOIN distributed_query_task t
        ON q.id = t.distributed_query_id
        ORDER BY t.id;
    """)

    results = map(DistributedQuery._make, cursor.fetchall())

    op.drop_constraint(u'distributed_query_result_distributed_query_task_id_fkey', 'distributed_query_result', type_='foreignkey')
    op.drop_column(u'distributed_query_result', 'distributed_query_task_id')

    op.drop_constraint(u'distributed_query_task_distributed_query_id_fkey', 'distributed_query_task', type_='foreignkey')
    op.drop_table(u'distributed_query')

    distributed_query = op.create_table('distributed_query',
        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('sql', sa.String(), nullable=False),
        sa.Column('timestamp', sa.DateTime(), nullable=True),
        sa.Column('not_before', sa.DateTime(), nullable=True),
        sa.Column('retrieved', sa.DateTime(), nullable=True),
        sa.Column('node_id', sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(['node_id'], ['node.id'], ),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('guid')
    )

    distributed_queries = [dict(
        guid=r.guid,
        status=r.status,
        sql=r.sql,
        timestamp=r.timestamp,
        not_before=r.not_before,
        retrieved=r.retrieved,
        node_id=r.node_id) for r in results]

    op.bulk_insert(distributed_query, distributed_queries)
    op.drop_table('distributed_query_task')
    ### end Alembic commands ###