<commit_before>from .job import Job


class Broker(object):

    def __init__(self, connector):
        self.connector = connector

    def __repr__(self):
        return 'Broker({connector})'.format(
            connector=type(self.connector).__name__
        )

    def add_job(self, job_class, *args, **kwargs):
        if not issubclass(job_class, Job):
            raise ValueError('task must be a subclass of Job')

        payload = self._gen_job_payload(job_class, args, kwargs)
        self.connector.enqueue(job_class.queue, payload)

    def jobs(self, queue_name):
        while True:
            payload = self.connector.dequeue(queue_name)

            if payload:
                yield payload

    def delete_job(self, job):
        self.connector.delete(job.queue, job.id)

    def set_retry_time(self, job, delay):
        self.connector.set_retry_time(job.queue, job.id, delay)

    def _gen_job_payload(self, job_class, args, kwargs):
        return {<commit_msg>Added timeout parameter to broker jobs()<commit_after>from .job import Job


class Broker(object):

    def __init__(self, connector):
        self.connector = connector

    def __repr__(self):
        return 'Broker({connector})'.format(
            connector=type(self.connector).__name__
        )

    def add_job(self, job_class, *args, **kwargs):
        if not issubclass(job_class, Job):
            raise ValueError('task must be a subclass of Job')

        payload = self._gen_job_payload(job_class, args, kwargs)
        self.connector.enqueue(job_class.queue, payload)

    def jobs(self, queue_name, timeout=20):
        while True:
            payload = self.connector.dequeue(queue_name, wait_time=timeout)

            if payload or timeout == 0:
                yield payload

    def delete_job(self, job):
        self.connector.delete(job.queue, job.id)

    def set_retry_time(self, job, delay):
        self.connector.set_retry_time(job.queue, job.id, delay)

    def _gen_job_payload(self, job_class, args, kwargs):
        return {
{'fraction_non_alphanumeric': 0.10294117647058823, 'fraction_numerical': 0.0014705882352941176, 'mean_word_length': 2.6446428571428573, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1306420', 'n_tokens_mistral': 625, 'n_tokens_neox': 602, 'n_words': 146}