def inspect_queue(queue, state, limit, bucket):
    """Show contents of a queue."""
    if not HAVE_BIN_LIBS:
        click.echo("missing required binary libs (lz4, msgpack)")
        return

    conn = worker.connection

    def job_row(j):
        if isinstance(j.args[0], basestring):
            account, bucket = j.args[0].split(':', 1)
        elif isinstance(j.args[0], dict):
            account, bucket = j.args[0]['name'], "set %d" % len(j.args[1])

        row = {
            'account': account,
            'bucket': bucket,
            # 'region': j.args[1]['region'],
            # 'size': j.args[1]['keycount'],
            'ttl': j.ttl,
            'enqueued': j.enqueued_at,
            'rtt': j.result_ttl,
            'timeout': j.timeout}

        if queue != "bucket-keyset-scan":
            row['args'] = j.args[2:]
        if state in ('running', 'failed', 'finished'):
            row['started'] = j.started_at
        if state in ('finished', 'failed'):
            row['ended'] = j.ended_at
        return row

    if state == 'running':
        registry_class = StartedJobRegistry
    elif state == 'pending':
        registry_class = Queue
    elif state == 'failed':
        registry_class = FailedQueue
    elif state == 'finished':
        registry_class = FinishedJobRegistry
    else:
        raise ValueError("invalid state: %s" % state)

    registry = registry_class(queue, connection=conn)
    records = []
    for jid in registry.get_job_ids():
        j = Job.fetch(jid, conn)
        if bucket:
            if j.args[1]['name'] != bucket:
                continue
        records.append(job_row(j))
        if len(records) == limit:
            break
    if records:
        click.echo(
            tabulate.tabulate(
                records,
                "keys",
                tablefmt='simple'))
    else:
        click.echo("no queue items found")
def inspect_queue ( queue , state , limit , bucket ) : if not HAVE_BIN_LIBS : click . echo ( "missing required binary libs (lz4, msgpack)" ) return conn = worker . connection def job_row ( j ) : if isinstance ( j . args [ 0 ] , basestring ) : account , bucket = j . args [ 0 ] . split ( ':' , 1 ) elif isinstance ( j . args [ 0 ] , dict ) : account , bucket = j . args [ 0 ] [ 'name' ] , "set %d" % len ( j . args [ 1 ] ) row = { 'account' : account , 'bucket' : bucket , 'ttl' : j . ttl , 'enqueued' : j . enqueued_at , 'rtt' : j . result_ttl , 'timeout' : j . timeout } if queue != "bucket-keyset-scan" : row [ 'args' ] = j . args [ 2 : ] if state in ( 'running' , 'failed' , 'finished' ) : row [ 'started' ] = j . started_at if state in ( 'finished' , 'failed' ) : row [ 'ended' ] = j . ended_at return row if state == 'running' : registry_class = StartedJobRegistry elif state == 'pending' : registry_class = Queue elif state == 'failed' : registry_class = FailedQueue elif state == 'finished' : registry_class = FinishedJobRegistry else : raise ValueError ( "invalid state: %s" % state ) registry = registry_class ( queue , connection = conn ) records = [ ] for jid in registry . get_job_ids ( ) : j = Job . fetch ( jid , conn ) if bucket : if j . args [ 1 ] [ 'name' ] != bucket : continue records . append ( job_row ( j ) ) if len ( records ) == limit : break if records : click . echo ( tabulate . tabulate ( records , "keys" , tablefmt = 'simple' ) ) else : click . echo ( "no queue items found" )
Show contents of a queue.