<commit_before># -*- coding: utf-8 -*-

"""Provides ``TaskPerformer``, a utility that aquires a task from the db,
  and performs it by making a POST request to the task's web hook url.
"""

__all__ = [
    'TaskPerformer',
]

import logging
logger = logging.getLogger(__name__)

import gevent
import requests

from torque import backoff
from torque import model

class TaskPerformer(object):
    def __init__(self, **kwargs):
        self.task_manager = kwargs.get('acquire_task', model.TaskManager())
        self.backoff_cls = kwargs.get('backoff', backoff.Backoff)
        self.post = kwargs.get('post', requests.post)
        self.sleep = kwargs.get('spawn', gevent.sleep)
        self.spawn = kwargs.get('spawn', gevent.spawn)
    
    def __call__(self, instruction, control_flag):
        """Parse the instruction to transactionally get-and-incr the task."""
        
        # Parse the instruction to transactionally get-and-incr the task.
        task_id, retry_count = map(int, instruction.split(':'))
        task_data = self.task_manager.acquire(task_id, retry_count)
        
        # Unpack the task data.
        url = task_data['url']
        body = task_data['body']
        timeout = task_data['timeout']
        headers = {'content-type': '{0}; charset={1}'.format(
                task_data['enctype'], task_data['charset'])}
        
        # Spawn a POST to the web hook in a greenlet -- so we can monitor
        # the control flag in case we want to exit whilst waiting.
        kwargs = dict(data=body, headers=headers, timeout=timeout)
        greenlet = self.spawn(self.post, url, **kwargs)
        
        # Wait for the request to complete, checking the greenlet's progress
        # increasingly less frequently.
        response = None
        delay = 0.1 # secs
        max_delay = 5 # secs
        backoff = self.backoff_cls(start_delay, max_value=max_delay)
        while control_flag.is_set():
            self.sleep(delay)
            if greenlet.ready():
                response = greenlet.value
                break
            delay = backoff.exponential(1.5) # 0.15, 0.225, 0.3375, 0.50625 ... 5
        
        # If we didn't get a response, or if the response was not successful,
        # exit gracefully, leaving the task to be retried in due course.
        if response is None or response.status_code > 200:
            current_retry_count = task_data['retry_count']
            self.task_manager.reschedule(task_id, current_retry_count)
        else: # Flag the task as complete.
            self.task_manager.complete(task_id)
    


"""<commit_msg>work: add comment above reschedule.<commit_after># -*- coding: utf-8 -*-

"""Provides ``TaskPerformer``, a utility that aquires a task from the db,
  and performs it by making a POST request to the task's web hook url.
"""

__all__ = [
    'TaskPerformer',
]

import logging
logger = logging.getLogger(__name__)

import gevent
import requests

from torque import backoff
from torque import model

class TaskPerformer(object):
    def __init__(self, **kwargs):
        self.task_manager = kwargs.get('acquire_task', model.TaskManager())
        self.backoff_cls = kwargs.get('backoff', backoff.Backoff)
        self.post = kwargs.get('post', requests.post)
        self.sleep = kwargs.get('spawn', gevent.sleep)
        self.spawn = kwargs.get('spawn', gevent.spawn)
    
    def __call__(self, instruction, control_flag):
        """Parse the instruction to transactionally get-and-incr the task."""
        
        # Parse the instruction to transactionally get-and-incr the task.
        task_id, retry_count = map(int, instruction.split(':'))
        task_data = self.task_manager.acquire(task_id, retry_count)
        
        # Unpack the task data.
        url = task_data['url']
        body = task_data['body']
        timeout = task_data['timeout']
        headers = {'content-type': '{0}; charset={1}'.format(
                task_data['enctype'], task_data['charset'])}
        
        # Spawn a POST to the web hook in a greenlet -- so we can monitor
        # the control flag in case we want to exit whilst waiting.
        kwargs = dict(data=body, headers=headers, timeout=timeout)
        greenlet = self.spawn(self.post, url, **kwargs)
        
        # Wait for the request to complete, checking the greenlet's progress
        # increasingly less frequently.
        response = None
        delay = 0.1 # secs
        max_delay = 5 # secs
        backoff = self.backoff_cls(start_delay, max_value=max_delay)
        while control_flag.is_set():
            self.sleep(delay)
            if greenlet.ready():
                response = greenlet.value
                break
            delay = backoff.exponential(1.5) # 0.15, 0.225, 0.3375, 0.50625 ... 5
        
        # If we didn't get a response, or if the response was not successful,
        # exit gracefully, leaving the task to be retried in due course.
        if response is None or response.status_code > 200:
            current_retry_count = task_data['retry_count']
            # Note that rescheduling *accelerates* the due date -- doing nothing
            # here would leave the task to be retried at the same delay, plus
            # the timeout.
            # XXX what we could also do here are:
            # - set a more informative status flag (even if only descriptive)
            # - noop if the greenlet request timed out
            self.task_manager.reschedule(task_id, current_retry_count)
        else:
            # Flag the task as complete.
            self.task_manager.complete(task_id)
    


"""
{'fraction_non_alphanumeric': 0.09926929246123685, 'fraction_numerical': 0.010693281055070397, 'mean_word_length': 2.3705705705705706, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '2614649', 'n_tokens_mistral': 1647, 'n_tokens_neox': 1593, 'n_words': 499}