arthur/jobs.py
def execute_perceval_job(backend, backend_args, qitems, task_id, category,
                         archive_args=None, max_retries=MAX_JOB_RETRIES):
    """Execute a Perceval job on RQ.

    The items fetched during the process will be stored in a
    Redis queue named `queue`.

    Setting the parameter `archive_path`, raw data will be stored
    with the archive manager. The contents from the archive can
    be retrieved setting the pameter `fetch_from_archive` to `True`,
    too. Take into account this behaviour will be only available
    when the backend supports the use of the archive. If archiving
    is not supported, an `AttributeError` exception will be raised.

    :param backend: backend to execute
    :param bakend_args: dict of arguments for running the backend
    :param qitems: name of the RQ queue used to store the items
    :param task_id: identifier of the task linked to this job
    :param category: category of the items to retrieve
    :param archive_args: archive arguments
    :param max_retries: maximum number of attempts this job can execute
        before failing

    :returns: a `JobResult` instance

    :raises NotFoundError: raised when the backend is not found
    :raises AttributeError: raised when archiving is not supported but
        any of the archive parameters were set
    """
    rq_job = rq.get_current_job()

    job = PercevalJob(rq_job.id, task_id, backend, category,
                      rq_job.connection, qitems)

    logger.debug("Running job #%s (task: %s) (%s) (cat:%s)",
                 job.job_id, task_id, backend, category)

    if not job.has_archiving() and archive_args:
        raise AttributeError("archive attributes set but archive is not supported")

    run_job = True
    resume = False
    failures = 0

    while run_job:
        try:
            job.run(backend_args, archive_args=archive_args, resume=resume)
        except AttributeError as e:
            raise e
        except Exception as e:
            logger.debug("Error running job %s (%s) - %s",
                         job.job_id, backend, str(e))
            failures += 1

            if not job.has_resuming() or failures >= max_retries:
                logger.error("Cancelling job #%s (task: %s) (%s)",
                             job.job_id, task_id, backend)
                raise e

            logger.warning("Resuming job #%s (task: %s) (%s) due to a failure (n %s, max %s)",
                           job.job_id, task_id, backend, failures, max_retries)
            resume = True
        else:
            # No failure, do not retry
            run_job = False

    result = job.result

    logger.debug("Job #%s (task: %s) completed (%s) - %s items (%s) fetched",
                 result.job_id, task_id, result.backend, str(result.nitems), result.category)

    return result
def execute_perceval_job(backend, backend_args, qitems, task_id, category,
                         archive_args=None, max_retries=MAX_JOB_RETRIES):
    rq_job = rq.get_current_job()

    job = PercevalJob(rq_job.id, task_id, backend, category,
                      rq_job.connection, qitems)

    logger.debug("Running job #%s (task: %s) (%s) (cat:%s)",
                 job.job_id, task_id, backend, category)

    if not job.has_archiving() and archive_args:
        raise AttributeError("archive attributes set but archive is not supported")

    run_job = True
    resume = False
    failures = 0

    while run_job:
        try:
            job.run(backend_args, archive_args=archive_args, resume=resume)
        except AttributeError as e:
            raise e
        except Exception as e:
            logger.debug("Error running job %s (%s) - %s",
                         job.job_id, backend, str(e))
            failures += 1

            if not job.has_resuming() or failures >= max_retries:
                logger.error("Cancelling job #%s (task: %s) (%s)",
                             job.job_id, task_id, backend)
                raise e

            logger.warning("Resuming job #%s (task: %s) (%s) due to a failure (n %s, max %s)",
                           job.job_id, task_id, backend, failures, max_retries)
            resume = True
        else:
            # No failure, do not retry
            run_job = False

    result = job.result

    logger.debug("Job #%s (task: %s) completed (%s) - %s items (%s) fetched",
                 result.job_id, task_id, result.backend, str(result.nitems), result.category)

    return result
def execute_perceval_job ( backend , backend_args , qitems , task_id , category , archive_args = None , max_retries = MAX_JOB_RETRIES ) : rq_job = rq . get_current_job ( ) job = PercevalJob ( rq_job . id , task_id , backend , category , rq_job . connection , qitems ) logger . debug ( "Running job #%s (task: %s) (%s) (cat:%s)" , job . job_id , task_id , backend , category ) if not job . has_archiving ( ) and archive_args : raise AttributeError ( "archive attributes set but archive is not supported" ) run_job = True resume = False failures = 0 while run_job : try : job . run ( backend_args , archive_args = archive_args , resume = resume ) except AttributeError as e : raise e except Exception as e : logger . debug ( "Error running job %s (%s) - %s" , job . job_id , backend , str ( e ) ) failures += 1 if not job . has_resuming ( ) or failures >= max_retries : logger . error ( "Cancelling job #%s (task: %s) (%s)" , job . job_id , task_id , backend ) raise e logger . warning ( "Resuming job #%s (task: %s) (%s) due to a failure (n %s, max %s)" , job . job_id , task_id , backend , failures , max_retries ) resume = True else : # No failure, do not retry run_job = False result = job . result logger . debug ( "Job #%s (task: %s) completed (%s) - %s items (%s) fetched" , result . job_id , task_id , result . backend , str ( result . nitems ) , result . category ) return result
Execute a Perceval job on RQ.

    The items fetched during the process will be stored in a
    Redis queue named `queue`.

    Setting the parameter `archive_path`, raw data will be stored
    with the archive manager. The contents from the archive can
    be retrieved setting the pameter `fetch_from_archive` to `True`,
    too. Take into account this behaviour will be only available
    when the backend supports the use of the archive. If archiving
    is not supported, an `AttributeError` exception will be raised.

    :param backend: backend to execute
    :param bakend_args: dict of arguments for running the backend
    :param qitems: name of the RQ queue used to store the items
    :param task_id: identifier of the task linked to this job
    :param category: category of the items to retrieve
    :param archive_args: archive arguments
    :param max_retries: maximum number of attempts this job can execute
        before failing

    :returns: a `JobResult` instance

    :raises NotFoundError: raised when the backend is not found
    :raises AttributeError: raised when archiving is not supported but
        any of the archive parameters were set
Execute a Perceval job on RQ .