<commit_before>"""celery.backends.amqp"""
from carrot.connection import DjangoBrokerConnection
from carrot.messaging import Consumer, Publisher
from celery.backends.base import BaseBackend

RESULTSTORE_EXCHANGE = "celres"


class Backend(BaseBackend):
    """AMQP backend. Publish results by sending messages to the broker
    using the task id as routing key.

    Note that results published using this backend is read-once only.
    After the result has been read, the result is deleted.

    """

    capabilities = ["ResultStore"]

    def __init__(self, *args, **kwargs):
        super(Backend, self).__init__(*args, **kwargs)
        self._cache = {}

    def _publisher_for_task_id(self, task_id, connection):
        routing_key = task_id.replace("-", "")
        return Publisher(connection, exchange=RESULTSTORE_EXCHANGE,
                         exchange_type="direct",
                         routing_key="%s" % routing_key)

    def _consumer_for_task_id(self, task_id, connection):
        routing_key = task_id.replace("-", "")
        return Consumer(connection, queue=task_id,
                        exchange=RESULTSTORE_EXCHANGE,
                        exchange_type="direct",
                        routing_key="%s" % routing_key)

    def store_result(self, task_id, result, status):
        """Mark task as done (executed)."""
        if status == "DONE":
            result = self.prepare_result(result)
        elif status == "FAILURE":
            result = self.prepare_exception(result)

        meta = {"task_id": task_id,
                "result": result,
                "status": status}

        connection = DjangoBrokerConnection()
        publisher = self._publisher_for_task_id(task_id, connection)
        consumer = self._consumer_for_task_id(task_id, connection)
        c.fetch()
        publisher.send(meta, serializer="pickle", immediate=False)
        publisher.close()
        connection.close()
                        
        return result

    def is_done(self, task_id):
        """Returns ``True`` if task with ``task_id`` has been executed."""
        return self.get_status(task_id) == "DONE"

    def get_status(self, task_id):
        """Get the status of a task."""
        return self._get_task_meta_for(task_id)["status"]

    def _get_task_meta_for(self, task_id):
        results = []

        def callback(message_data, message):
            results.append(message_data)
            message.ack()

        routing_key = task_id.replace("-", "")

        connection = DjangoBrokerConnection()
        consumer = self._consumer_for_task_id(task_id, connection)
        consumer.register_callback(callback)
        
        try:
            consumer.iterconsume().next()
        finally:
            consumer.close()
            connection.close()

        return results[0]

    def get_result(self, task_id):
        """Get the result for a task."""
        result = self._get_task_meta_for(task_id) 
        if result["status"] == "FAILURE":
            return self.exception_to_python(result["result"])
        else:
            return result["result"]

    def cleanup(self):
        """Delete expired metadata."""
        pass<commit_msg>AMQP result-store backend is now working (and it's fast!)<commit_after>"""celery.backends.amqp"""
from carrot.connection import DjangoBrokerConnection
from carrot.messaging import Consumer, Publisher
from celery.backends.base import BaseBackend

RESULTSTORE_EXCHANGE = "celeryresults"


class Backend(BaseBackend):
    """AMQP backend. Publish results by sending messages to the broker
    using the task id as routing key.

    **NOTE:** Results published using this backend is read-once only.
    After the result has been read, the result is deleted. (however, it's
    still cached locally by the backend instance).

    """

    capabilities = ["ResultStore"]

    def __init__(self, *args, **kwargs):
        super(Backend, self).__init__(*args, **kwargs)
        self.connection = DjangoBrokerConnection()
        self._cache = {}

    def _declare_queue(self, task_id, connection):
        routing_key = task_id.replace("-", "")
        backend = connection.create_backend()
        backend.queue_declare(queue=routing_key, durable=True,
                                exclusive=False, auto_delete=True)
        backend.exchange_declare(exchange=RESULTSTORE_EXCHANGE,
                                 type="direct",
                                 durable=True,
                                 auto_delete=False)
        backend.queue_bind(queue=routing_key, exchange=RESULTSTORE_EXCHANGE,
                           routing_key=routing_key)
        backend.close()

    def _publisher_for_task_id(self, task_id, connection):
        routing_key = task_id.replace("-", "")
        self._declare_queue(task_id, connection)
        p = Publisher(connection, exchange=RESULTSTORE_EXCHANGE,
                      exchange_type="direct",
                      routing_key=routing_key)
        return p

    def _consumer_for_task_id(self, task_id, connection):
        routing_key = task_id.replace("-", "")
        self._declare_queue(task_id, connection)
        return Consumer(connection, queue=routing_key,
                        exchange=RESULTSTORE_EXCHANGE,
                        exchange_type="direct",
                        no_ack=False, auto_ack=False,
                        auto_delete=True,
                        routing_key=routing_key)

    def store_result(self, task_id, result, status):
        """Send task return value and status."""
        if status == "DONE":
            result = self.prepare_result(result)
        elif status == "FAILURE":
            result = self.prepare_exception(result)


        meta = {"task_id": task_id,
                "result": result,
                "status": status}

        connection = self.connection
        publisher = self._publisher_for_task_id(task_id, connection)
        publisher.send(meta, serializer="pickle")
        publisher.close()

        print("SENT %s RESULT: %s TO %s" % (
            status, result, task_id.replace("-", "")))
        return result

    def is_done(self, task_id):
        """Returns ``True`` if task with ``task_id`` has been executed."""
        return self.get_status(task_id) == "DONE"

    def get_status(self, task_id):
        """Get the status of a task."""
        return self._get_task_meta_for(task_id)["status"]

    def _get_task_meta_for(self, task_id):

        if task_id in self._cache:
            return self._cache[task_id]

        results = []

        def callback(message_data, message):
            results.append(message_data)
            message.ack()

        routing_key = task_id.replace("-", "")

        connection = self.connection
        consumer = self._consumer_for_task_id(task_id, connection)
        consumer.register_callback(callback)
        
        try:
            consumer.iterconsume().next()
        finally:
            consumer.backend.channel.queue_delete(routing_key)
            consumer.close()

        self._cache[task_id] = results[0]
        return results[0]

    def get_result(self, task_id):
        """Get the result for a task."""
        result = self._get_task_meta_for(task_id) 
        if result["status"] == "FAILURE":
            return self.exception_to_python(result["result"])
        else:
            return result["result"]
{'fraction_non_alphanumeric': 0.10408520965349873, 'fraction_numerical': 0.00040447620331670486, 'mean_word_length': 2.293960923623446, 'pattern_counts': {'":': 12, '<': 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': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5074170', 'n_tokens_mistral': 2073, 'n_tokens_neox': 2014, 'n_words': 466}