<commit_before>import pika
from django.conf import settings


def push_to_queue(queue_name, qitem=None):
    '''
    Publishes qitem (serialized data) to a specified queue.
    Returns the number of outstanding messages in specified queue
    '''
    queue_name = str(queue_name)  # Important: queue_name cannot be unicode!

    if queue_name == 'null':
        return 0
    credentials = pika.PlainCredentials(settings.RABBITMQ_USER,
                                            settings.RABBITMQ_PASS)
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        heartbeat_interval=5, credentials=credentials,
        host=settings.RABBIT_HOST))
    channel = connection.channel()
    q = channel.queue_declare(queue=queue_name, durable=True)
    if qitem is not None:
        channel.basic_publish(exchange='',
            routing_key=queue_name,
            body=qitem,
            properties=pika.BasicProperties(delivery_mode=2),
            )
    connection.close()
    return q.method.message_count


def get_queue_length(queue_name):
    """
    push_to_queue is not a great name for a function
    that returns the queue length, so make an alias
    """
    return push_to_queue(queue_name)<commit_msg>Add retries to push_to_queue producer<commit_after>import time

from django.conf import settings

import pika


MAX_RETRIES = 5
RETRY_TIMEOUT = 0.5


def push_to_queue(queue_name, qitem=None):
    """
    Publishes qitem (serialized data) to a specified queue.
    Returns the number of outstanding messages in specified queue
    """
    queue_name = str(queue_name)  # Important: queue_name cannot be unicode!

    if queue_name == 'null':
        return 0

    credentials = pika.PlainCredentials(settings.RABBITMQ_USER,
                                        settings.RABBITMQ_PASS)

    parameters = pika.ConnectionParameters(heartbeat_interval=5,
                                           credentials=credentials,
                                           host=settings.RABBIT_HOST)

    retries = 0
    while True:
        try:
            connection = pika.BlockingConnection(parameters)
        except pika.exceptions.AMQPConnectionError as e:
            retries += 1
            if retries >= MAX_RETRIES:
                raise e
            time.sleep(RETRY_TIMEOUT)
            continue
        else:
            break

    channel = connection.channel()

    queue = channel.queue_declare(queue=queue_name, durable=True)

    if qitem is not None:
        channel.basic_publish(exchange='',
                              routing_key=queue_name,
                              body=qitem,
                              properties=pika.BasicProperties(delivery_mode=2))

    connection.close()
    return queue.method.message_count


def get_queue_length(queue_name):
    """
    push_to_queue is not a great name for a function
    that returns the queue length, so make an alias
    """
    return push_to_queue(queue_name)
{'fraction_non_alphanumeric': 0.07015177065767285, 'fraction_numerical': 0.003709949409780776, 'mean_word_length': 2.273730684326711, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3662792', 'n_tokens_mistral': 827, 'n_tokens_neox': 795, 'n_words': 202}