<commit_before># The MIT License (MIT)
#
# Copyright (c) 2014 <NAME>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#

"""Common worker application module. Provides the interface with which
producers and executors see the task system.

.. class:: _TaskCaller

   This object represents a task that may be called synchronously in the
   current process, or asynchronously by a separate worker process.

   .. method:: delay(*args, **kwargs)

      Convenience method for calling :meth:`.apply_async` in a more
      traditional fashion.

      :param args: Positional arguments passed in to the task function.
      :type args: positional arguments
      :param kwargs: Keyword arguments passed in to the task function.
      :type kwargs: keyword arguments
      :returns: A way to retrieve the result of the task's execution when it
                is ready.
      :rtype: :class:`AsyncResult`

   .. method:: apply_async(args, kwargs=None, correlation_id=None,
                           routing_key=None, send_result=False)

      Triggers the asynchronous execution of a task by a separate worker
      process. Tasks will be executed in the order they are triggered, so a
      task may not be executed immediately.

      :param args: Positional arguments to be passed in to the task function.
      :type args: list
      :param kwargs: Keyword arguments to be passed in to the task function.
      :type kwargs: dict
      :param correlation_id: Force the message to use a specific correlation
                             identifier string.
      :type correlation_id: str
      :param routing_key: Override the default AMQP routing key for the task
                          with the given string.
      :type routing_key: str
      :param send_result: Create a temporary result queue and request that the
                          worker publish the task's result to it.
      :type send_result: bool
      :returns: If ``send_result`` is True, used retrieve the result of the
                task's execution when it is ready. Otherwise, returns ``None``.
      :rtype: :class:`AsyncResult`

   .. method:: apply(args, kwargs=None, correlation_id=None)

      Calls the task function synchronously in the current process, returning
      the result. This is equivalent to a normal function call, except
      start and finish events are logged.

      :param args: Arguments passed directly into the function.
      :type args: list
      :param kwargs: Arguments passed directly into the function.
      :type kwargs: dict
      :param correlation_id: The task being executed had this string as its
                             correlation identifier.
      :type correlation_id: str
      :returns: The return value of the function call.

   .. method:: __call__(*args, **kwargs)

      Equivalent to :meth:`.apply` passing in ``args`` and ``kwargs``.

      :param args: Arguments passed directly into the function.
      :type args: positional arguments
      :param kwargs: Arguments passed directly into the function.
      :type kwargs: keyword arguments
      :returns: The return value of the function call.

"""

from __future__ import absolute_import

import time
import json
import errno
import logging
from uuid import uuid4
from socket import timeout as socket_timeout, error as socket_error
from multiprocessing import TimeoutError

from six.moves import cPickle
import amqp

from .amqp import AmqpConnection

__all__ = ['AsyncResult', 'WorkerApplication', 'routing_info']


logger = logging.getLogger('provoke.tasks')


class AsyncResult(object):
    """Used to query for the results of an asynchronous operation. Compatible
    with the builtin :py:class:`multiprocessing.pool.AsyncResult` class
    interface.

    :param correlation_id: The ID used to identify tasks and correlate their
                           request and response messages.
    :type correlation_id: str

    """

    def __init__(self, correlation_id):
        super(AsyncResult, self).__init__()
        self.correlation_id = correlation_id

    @property
    def result_queue(self):
        """The name of the queue where results will be published."""
        return 'result_{0}'.format(self.correlation_id)

    @property
    def args(self):
        """If the result is available, these will be a tuple of the original
        positional and keyword arguments sent with the request.

        """
        if self.ready():
            return (self._result['args'], self._result['kwargs'])
        raise AttributeError<commit_msg>allow alternative result queue prefix<commit_after># The MIT License (MIT)
#
# Copyright (c) 2014 <NAME>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#

"""Common worker application module. Provides the interface with which
producers and executors see the task system.

.. class:: _TaskCaller

   This object represents a task that may be called synchronously in the
   current process, or asynchronously by a separate worker process.

   .. method:: delay(*args, **kwargs)

      Convenience method for calling :meth:`.apply_async` in a more
      traditional fashion.

      :param args: Positional arguments passed in to the task function.
      :type args: positional arguments
      :param kwargs: Keyword arguments passed in to the task function.
      :type kwargs: keyword arguments
      :returns: A way to retrieve the result of the task's execution when it
                is ready.
      :rtype: :class:`AsyncResult`

   .. method:: apply_async(args, kwargs=None, correlation_id=None,
                           routing_key=None, send_result=False)

      Triggers the asynchronous execution of a task by a separate worker
      process. Tasks will be executed in the order they are triggered, so a
      task may not be executed immediately.

      :param args: Positional arguments to be passed in to the task function.
      :type args: list
      :param kwargs: Keyword arguments to be passed in to the task function.
      :type kwargs: dict
      :param correlation_id: Force the message to use a specific correlation
                             identifier string.
      :type correlation_id: str
      :param routing_key: Override the default AMQP routing key for the task
                          with the given string.
      :type routing_key: str
      :param send_result: Create a result queue and request that the worker
                          publish the task's result to it. Use a `Queue TTL
                          <https://www.rabbitmq.com/ttl.html#queue-ttl>`_
                          policy to expire these queues.
      :type send_result: bool
      :returns: If ``send_result`` is True, used retrieve the result of the
                task's execution when it is ready. Otherwise, returns ``None``.
      :rtype: :class:`AsyncResult`

   .. method:: apply(args, kwargs=None, correlation_id=None)

      Calls the task function synchronously in the current process, returning
      the result. This is equivalent to a normal function call, except
      start and finish events are logged.

      :param args: Arguments passed directly into the function.
      :type args: list
      :param kwargs: Arguments passed directly into the function.
      :type kwargs: dict
      :param correlation_id: The task being executed had this string as its
                             correlation identifier.
      :type correlation_id: str
      :returns: The return value of the function call.

   .. method:: __call__(*args, **kwargs)

      Equivalent to :meth:`.apply` passing in ``args`` and ``kwargs``.

      :param args: Arguments passed directly into the function.
      :type args: positional arguments
      :param kwargs: Arguments passed directly into the function.
      :type kwargs: keyword arguments
      :returns: The return value of the function call.

"""

from __future__ import absolute_import

import time
import json
import errno
import logging
from uuid import uuid4
from socket import timeout as socket_timeout, error as socket_error
from multiprocessing import TimeoutError

from six.moves import cPickle
import amqp

from .amqp import AmqpConnection

__all__ = ['AsyncResult', 'WorkerApplication', 'routing_info']


logger = logging.getLogger('provoke.tasks')


class AsyncResult(object):
    """Used to query for the results of an asynchronous operation. Compatible
    with the builtin :py:class:`multiprocessing.pool.AsyncResult` class
    interface.

    :param correlation_id: The ID used to identify tasks and correlate their
                           request and response messages.
    :type correlation_id: str

    """

    #: The prefix given to the result queue.
    result_queue_prefix = 'result_'

    def __init__(self, correlation_id):
        super(AsyncResult, self).__init__()
        self.correlation_id = correlation_id

    @property
    def result_queue(self):
        """The name of the queue where results will be published."""
        return '{0}{1}'.format(self.result_queue_prefix, self.correlation_id)

    @property
    def args(self):
        """If the result is available, these will be a tuple of the original
        positional and keyword arguments sent with the request.

        """
        if self.ready():
            return (self._result['args'], self._result['kwargs'])
        raise AttributeError
{'fraction_non_alphanumeric': 0.06395348837209303, 'fraction_numerical': 0.0011451726568005639, 'mean_word_length': 3.0216082182075805, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 6, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, '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': '7507153', 'n_tokens_mistral': 2852, 'n_tokens_neox': 2742, 'n_words': 1352}