(app_name, job_id, _force=False,
                  _reset_descendants=True, _ignore_if_queued=False)
return True
Queue a new task if it isn't already in the queue.

    This is slow if the queue is large.  Use carefully!
    Potentially unsafe if decendant tasks are already running/queued.

    `_force` (bool) If True, add to queue regardless of whether job_id
        is previously queued
    `_reset_descendants` (bool)  If True, recurse through descendants and reset
        their states to pending
    `_ignore_if_queued` (bool)  If the job_id is previously queued,
        don't re-queue it, but possibly _reset_descendants.
        If False, just raise an error

    In an atomic transaction (ie atomic to this job_id),
    First, validate that the job isn't already queued.
    Then, queue the task
Queue a new task if it isn't already in the queue.
Queue a new task if it isn t already in the queue .
def readd_subtask(app_name, job_id, _force=False,
                  _reset_descendants=True, _ignore_if_queued=False):
    """
    Queue a new task if it isn't already in the queue.

    This is slow if the queue is large.  Use carefully!
    Potentially unsafe if decendant tasks are already running/queued.

    `_force` (bool) If True, add to queue regardless of whether job_id
        is previously queued
    `_reset_descendants` (bool)  If True, recurse through descendants and reset
        their states to pending
    `_ignore_if_queued` (bool)  If the job_id is previously queued,
        don't re-queue it, but possibly _reset_descendants.
        If False, just raise an error

    In an atomic transaction (ie atomic to this job_id),
    First, validate that the job isn't already queued.
    Then, queue the task
    """
    qbcli = shared.get_qbclient()
    # obtain lock
    try:
        if _ignore_if_queued:  # don't bother waiting for an add lock.
            # Assume that ignore_if_queued == True implies the task might get
            # added to the queue while it is still in the queue.
            #  At this point,
            # we assume it makes no difference whether the task
            # is re-added once
            # or more.  Since multiple processes might be trying to re-add at
            # the same time, just let one of them do it and, at least in this
            # section, raise a failure for the others.
            lock = obtain_add_lock(
                app_name, job_id, blocking=False, raise_on_error=True)
        else:
            lock = obtain_add_lock(
                app_name, job_id, blocking=False, raise_on_error=True)
    except exceptions.CouldNotObtainLock:
        # call maybe_add_subtask(...) and return
        added = maybe_add_subtask(app_name, job_id)
        if not added:
            raise exceptions.CodeError(
                "wtf?  If I can't obtain a lock on a job_id, then I should"
                " be able to maybe_add_subtask.")
        return added
    except exceptions.LockAlreadyAcquired:
        # here's why something else might hold an add lock on the object:
        # - task currently being added by another process
        # - process that was performing above op died a few seconds ago
        raise
    assert lock, "Code Error: should have the add lock at this point"
    try:
        if _force:
            queued = False
        else:
            queued = qbcli.LockingQueue(app_name).is_queued(job_id)
            if queued and _ignore_if_queued:
                log.debug(
                    "Job already queued!  We'll handle this properly."
                    " You may have entered this state"
                    " because you manually re-added a child and parent task"
                    " and then completed the parent task first.  This could"
                    " also happen if, via bubble up, a child queued parents"
                    " and then a parent just tried, via"
                    " bubble down, to queue one of children.", extra=dict(
                        app_name=app_name, job_id=job_id)
                )
            elif queued:
                raise exceptions.JobAlreadyQueued(
                    "%s %s" % (app_name, job_id))

        if _reset_descendants:
            # all child tasks will also get re-executed
            _recursively_reset_child_task_state(app_name, job_id)

        if not queued:
            _queue(app_name, job_id)
    finally:
        lock.release()
    return True
def readd_subtask ( app_name , job_id , _force = False , _reset_descendants = True , _ignore_if_queued = False ) : qbcli = shared . get_qbclient ( ) # obtain lock try : if _ignore_if_queued : # don't bother waiting for an add lock. # Assume that ignore_if_queued == True implies the task might get # added to the queue while it is still in the queue. #  At this point, # we assume it makes no difference whether the task # is re-added once # or more.  Since multiple processes might be trying to re-add at # the same time, just let one of them do it and, at least in this # section, raise a failure for the others. lock = obtain_add_lock ( app_name , job_id , blocking = False , raise_on_error = True ) else : lock = obtain_add_lock ( app_name , job_id , blocking = False , raise_on_error = True ) except exceptions . CouldNotObtainLock : # call maybe_add_subtask(...) and return added = maybe_add_subtask ( app_name , job_id ) if not added : raise exceptions . CodeError ( "wtf?  If I can't obtain a lock on a job_id, then I should" " be able to maybe_add_subtask." ) return added except exceptions . LockAlreadyAcquired : # here's why something else might hold an add lock on the object: # - task currently being added by another process # - process that was performing above op died a few seconds ago raise assert lock , "Code Error: should have the add lock at this point" try : if _force : queued = False else : queued = qbcli . LockingQueue ( app_name ) . is_queued ( job_id ) if queued and _ignore_if_queued : log . debug ( "Job already queued!  We'll handle this properly." " You may have entered this state" " because you manually re-added a child and parent task" " and then completed the parent task first.  This could" " also happen if, via bubble up, a child queued parents" " and then a parent just tried, via" " bubble down, to queue one of children." , extra = dict ( app_name = app_name , job_id = job_id ) ) elif queued : raise exceptions . JobAlreadyQueued ( "%s %s" % ( app_name , job_id ) ) if _reset_descendants : # all child tasks will also get re-executed _recursively_reset_child_task_state ( app_name , job_id ) if not queued : _queue ( app_name , job_id ) finally : lock . release ( ) return True