<commit_before>#!/usr/bin/env python 
import os, sys, time, json
import requests
import logging
import traceback

import pika

logging.basicConfig(level=logging.INFO, format='%(asctime)s | %(levelname)s | %(message)s')

RETRY_DELAY = 30000 # Warning: do not change this if MQ has the queue instantiated already.

TEST_NAME = 'test.queue_trigger_pattern'    
TEST_EXCHANGES = (('notify.retail.sale.Sale.updated',None),('notify.retail.sale.Sale.created',{'store':'Test Store 1'}))


class RequeableError( Exception ):
    pass
    
    
class FatalError( Exception ):
    pass

    
class QueueTriggerPattern():
    """
    Boiler plate of queue triggered microservices. Should be inherited and 'worker' function overriden.
    
    Provided exchanges is a list of tuples. Each tuple contains an exchange name and exchange args dictionary (or None)
    Args example: {'ham': 'good', 'x-match':'any'}
    """
   
    def __init__(self, microservice_name, exchanges, queue_host, queue_user, queue_pass):
        self.microservice_name = microservice_name
        self.exchanges = exchanges
        self.retry_ready_exchnage = 'microservice.' + microservice_name + '.retry_ready'
        self.retry_exchange = 'microservice.' + microservice_name + '.retry'
        self.queue_name = 'microservice.' + microservice_name
        self.retry_queue_name = 'microservice.' + microservice_name + ".retry"
    
        self.queue_host = queue_host
        self.queue_user = queue_user
        self.queue_pass = queue_pass
    
    
    def worker(self, message_json):
        """
        Function as a placeholder to be overriden which is called for each message received on the subscribed queues
        """
        print( "Should override worker and do something here" )
    
    
    def request(self, url, user, password, error_message):
        """
        A wrapper around the requests get function which will call CBE and load the json response.
        """
        response = requests.get(url, auth=(user, password))
        if response.status_code >= 500 or response.status_code in (401,403):
            # Retryable errors which can be requeued
            logging.warning( "Retryable " + error_message + "{}".format(response.content) )
            raise RequeableError("Get {} returned: {}".format(url, response.status_code))
        elif response.status_code != 200:   
            # Fatal errors which can't be retried
            logging.error( "Fatal " + error_message + "{}".format(response.content) )
            raise FatalError("Get {} returned: {}".format(url, response.status_code))<commit_msg>Checked for json dict in request before calling GET<commit_after>#!/usr/bin/env python 
import os, sys, time, json
import requests
import logging
import traceback

import pika

logging.basicConfig(level=logging.INFO, format='%(asctime)s | %(levelname)s | %(message)s')

RETRY_DELAY = 30000 # Warning: do not change this if MQ has the queue instantiated already.

TEST_NAME = 'test.queue_trigger_pattern'    
TEST_EXCHANGES = (('notify.retail.sale.Sale.updated',None),('notify.retail.sale.Sale.created',{'store':'Test Store 1'}))


class RequeableError( Exception ):
    pass
    
    
class FatalError( Exception ):
    pass

    
class QueueTriggerPattern():
    """
    Boiler plate of queue triggered microservices. Should be inherited and 'worker' function overriden.
    
    Provided exchanges is a list of tuples. Each tuple contains an exchange name and exchange args dictionary (or None)
    Args example: {'ham': 'good', 'x-match':'any'}
    """
   
    def __init__(self, microservice_name, exchanges, queue_host, queue_user, queue_pass):
        self.microservice_name = microservice_name
        self.exchanges = exchanges
        self.retry_ready_exchnage = 'microservice.' + microservice_name + '.retry_ready'
        self.retry_exchange = 'microservice.' + microservice_name + '.retry'
        self.queue_name = 'microservice.' + microservice_name
        self.retry_queue_name = 'microservice.' + microservice_name + ".retry"
    
        self.queue_host = queue_host
        self.queue_user = queue_user
        self.queue_pass = queue_pass
    
    
    def worker(self, message_json):
        """
        Function as a placeholder to be overriden which is called for each message received on the subscribed queues
        """
        print( "Should override worker and do something here" )
    
    
    def request(self, url, user, password, error_message):
        """
        A wrapper around the requests get function which will call CBE and load the json response.
        """
        if type(url) is dict:
            logging.info( "QueueTriggerPattern request called to get url but url is already dict. Assuming loaded json so returning input." )
            return url
        response = requests.get(url, auth=(user, password))
        if response.status_code >= 500 or response.status_code in (401,403):
            # Retryable errors which can be requeued
            logging.warning( "Retryable " + error_message + "{}".format(response.content) )
            raise RequeableError("Get {} returned: {}".format(url, response.status_code))
        elif response.status_code != 200:
            # Fatal errors which can't be retried
            logging.error( "Fatal " + error_message + "{}".format(response.content) )
            raise FatalError("Get {} returned: {}".format(url, response.status_code))
{'fraction_non_alphanumeric': 0.10084652189915348, 'fraction_numerical': 0.006624953993375046, 'mean_word_length': 3.06203288490284, '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': '5556117', 'n_tokens_mistral': 1576, 'n_tokens_neox': 1516, 'n_words': 457}