<commit_before>import socket

from kombu import Connection, Queue, Consumer
from kombu.utils import nested
from kombu.pools import connections

from microservices.utils import get_logger

_logger = get_logger('Microservices queues')

class Rule(object):
    """Rule"""

    def __init__(self, name, handler, **options):
        """Initialization

        :param name: name of queue
        :param handler: handle for queue
        """
        self.handler = handler
        self.name = name
        self.options = options

    def callback(self, body, message):
        _logger.debug('Received data from %s' % (self.name, ))
        self.handler(body, HandlerContext(message, self))
        message.ack()

class HandlerContext(object):
    """Context for handler function"""

    def __init__(self, message, rule):
        """Initialization

        :param message: original message from kombu
        :type message: kombu.Message
        :param rule: rule object
        :type rule: Rule
        """
        self.message = message
        self.rule = rule

class Microservice(object):
    """Microservice for queues"""

    connection = 'amqp:///'

    def __init__(self, connection='amqp:///', logger=None, timeout=10):
        """Initialization

        :param connection: connection for queues broker
        :type connection: str, None, dict or Connection
        :param logger: logging instance
        :type logger: Logger
        :param timeout: sleeping for loop, default = 0.1
        :type timeout: None, int or float
        """
        if logger is None:
            logger = _logger

        self.logger = logger
        self.connection = self._get_connection(connection)
        self.timeout = timeout
        self.consumers = []

    def _get_connection(self, connection):
        """Create connection strategy

        :param connection: connection for broker
        :type connection: str, None, kombu.connections.Connection, dict
        :return: instance of kombu.connections.Connection
        :rtype: Connection
        """
        if not connection:
            connection = self.connection

        if isinstance(connection, Connection):
            return connection

        if isinstance(connection, str):
            connection = {'hostname': connection}

        return Connection(**connection)

    def add_queue_rule(self, handler, name, **kwargs):
        """Add queue rule to Microservice

        :param handler: function for handling messages
        :type handler: callable object
        :param name: name of queue
        :type name: str
        """

        with connections[self.connection].acquire() as conn:
            rule = Rule(name, handler, **kwargs)
            consumer = Consumer(conn, queues=[Queue(rule.name)], callbacks=[rule.callback], auto_declare=True)
            self.consumers.append(consumer)

    def queue(self, name, **kwargs):
        """Decorator for handler function

        >>>app = Microservice()
        >>>
        >>>@app.queue('queue')
        >>>def function(payload, context):
        >>>    pass

        :param name: name of queue
        :type name: str
        """
        def decorator(f):
            self.add_queue_rule(f, name, **kwargs)
            return f

        return decorator

    def drain_events(self, infinity=True):
        with connections[self.connection].acquire() as conn:
            with nested(*self.consumers):
                while True:
                    try:
                        conn.drain_events(timeout=self.timeout)
                    except socket.timeout:
                        if not infinity:
                            return

    def run(self, debug=False):
        """Run microservice in loop, where handle connections

        :param debug: enable/disable debug mode
        :type debug: bool
        """
        if debug:
            from microservices.utils import set_logging

            set_logging('DEBUG')
        self.drain_events(infinity=True)
<commit_msg>safe handling exceptions in queue service<commit_after>import socket

from kombu import Connection, Queue, Consumer
from kombu.utils import nested
from kombu.pools import connections

from microservices.utils import get_logger

_logger = get_logger('Microservices queues')

class Rule(object):
    """Rule"""

    def __init__(self, name, handler, **options):
        """Initialization

        :param name: name of queue
        :param handler: handle for queue
        """
        self.handler = handler
        self.name = name
        self.options = options

    def callback(self, body, message):
        _logger.debug('Received data from %s' % (self.name, ))
        self.handler(body, HandlerContext(message, self))
        message.ack()

class HandlerContext(object):
    """Context for handler function"""

    def __init__(self, message, rule):
        """Initialization

        :param message: original message from kombu
        :type message: kombu.Message
        :param rule: rule object
        :type rule: Rule
        """
        self.message = message
        self.rule = rule

class Microservice(object):
    """Microservice for queues"""

    connection = 'amqp:///'

    def __init__(self, connection='amqp:///', logger=None, timeout=10):
        """Initialization

        :param connection: connection for queues broker
        :type connection: str, None, dict or Connection
        :param logger: logging instance
        :type logger: Logger
        :param timeout: sleeping for loop, default = 0.1
        :type timeout: None, int or float
        """
        if logger is None:
            logger = _logger

        self.logger = logger
        self.connection = self._get_connection(connection)
        self.timeout = timeout
        self.consumers = []

    def _get_connection(self, connection):
        """Create connection strategy

        :param connection: connection for broker
        :type connection: str, None, kombu.connections.Connection, dict
        :return: instance of kombu.connections.Connection
        :rtype: Connection
        """
        if not connection:
            connection = self.connection

        if isinstance(connection, Connection):
            return connection

        if isinstance(connection, str):
            connection = {'hostname': connection}

        return Connection(**connection)

    def add_queue_rule(self, handler, name, **kwargs):
        """Add queue rule to Microservice

        :param handler: function for handling messages
        :type handler: callable object
        :param name: name of queue
        :type name: str
        """

        with connections[self.connection].acquire() as conn:
            rule = Rule(name, handler, **kwargs)
            consumer = Consumer(conn, queues=[Queue(rule.name)], callbacks=[rule.callback], auto_declare=True)
            self.consumers.append(consumer)

    def queue(self, name, **kwargs):
        """Decorator for handler function

        >>>app = Microservice()
        >>>
        >>>@app.queue('queue')
        >>>def function(payload, context):
        >>>    pass

        :param name: name of queue
        :type name: str
        """
        def decorator(f):
            self.add_queue_rule(f, name, **kwargs)
            return f

        return decorator

    def drain_events(self, infinity=True):
        with connections[self.connection].acquire() as conn:
            with nested(*self.consumers):
                while True:
                    try:
                        conn.drain_events(timeout=self.timeout)
                    except socket.timeout:
                        if not infinity:
                            return
                    except Exception as e:
                        self.logger.exception(e)

    def run(self, debug=False):
        """Run microservice in loop, where handle connections

        :param debug: enable/disable debug mode
        :type debug: bool
        """
        if debug:
            from microservices.utils import set_logging

            set_logging('DEBUG')
        self.drain_events(infinity=True)

{'fraction_non_alphanumeric': 0.09605425400739827, 'fraction_numerical': 0.0009864364981504315, 'mean_word_length': 2.104094910065059, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 33, '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': '5606030', 'n_tokens_mistral': 2075, 'n_tokens_neox': 1980, 'n_words': 632}