import json import re import uuid import requests from datetime import datetime, timedelta from utilities.schema_loader import s3_upload_service from key_resolver import KeyResolver from db.database_operations import DatabaseOperations from static_functions import StaticFunctions from logger_module.package_logger import print_log from exceptions.skipIterationException import callSkipFunction, skipItertionException class Operations: @classmethod def perform_operation(cls, operation, request_body): """ Helps execute the operation such as setting data, returning response, get length of data, etc :param operation: the operation dict block with all the details related to the operation :param request_body: request body received :return: sets data at global level, returns response or 1 """ try: from helper_functions import set_data if operation.get('operation') == 'returnResponse': if operation.get('return_code') or operation.get('message'): set_data('response', operation) return {'return_code': operation.get('return_code'), 'message': operation.get('message'), 'cf_code': operation.get('CF_code')} if operation.get('operation') == 'setData': set_value = KeyResolver().resolve_key(operation.get('value'), request_body) if set_value == 'UUID': set_value = str(uuid.uuid4()) value = set_data(operation.get('stateKey'), set_value) return value if operation.get('operation') == 'returnData': return_value = KeyResolver().resolve_key(operation.get('resultKey'), request_body) return return_value if operation.get('operation') == 'mapData': enum_dict = operation.get('enum') resolved_value = KeyResolver().resolve_key(operation.get('key'), request_body) enum_value = enum_dict.get(str(resolved_value)) value = set_data(operation.get('stateKey'), enum_value) return value if operation.get('operation') == 'length': value = KeyResolver().resolve_key(str(operation.get('key')), request_body) if isinstance(value, list): return set_data(operation.get('stateKey'), len(value)) else: return set_data(operation.get('stateKey'), 0 if not value else len(value)) if operation.get('operation') == 'concat': values = operation.get('values') for idx, items in enumerate(values): values[idx] = KeyResolver().resolve_key(items, request_body) number = KeyResolver().resolve_key(operation.get("number", 1), request_body) str_values = [str(i) for i in values] update_str = "".join(str_values) update_values = update_str * number if operation.get('resultKey'): return set_data(operation.get('resultKey'), update_values, True) elif operation.get('stateKey'): return set_data(operation.get('stateKey'), update_values) if operation.get('operation') == 'query': status, result = DatabaseOperations().execute(operation, request_body) return_value = False if operation.get('stateKey') or operation.get('auto_id') else True if not return_value: cols = operation.get('columns') if operation.get('queryType') == 'insert' and operation.get('auto_id'): return set_data(f'{operation.get("table")}_{operation.get("auto_id")}', result, return_value, cols) elif operation.get('queryType') in ['select', 'raw']: return set_data(operation.get('stateKey'), result, return_value, cols) return 1 if operation.get('operation') == 's3Upload': upload_content = KeyResolver().resolve_key(operation.get('contentKey'), request_body) path = KeyResolver().resolve_key(operation.get('path'), request_body) status = s3_upload_service(upload_content, path) if operation.get('resultKey'): return set_data(operation.get('resultKey'), status, True) if operation.get('stateKey'): return set_data(operation.get('stateKey'), status) if operation.get('operation') == 'calculate': expression = operation.get('value') expression_list = expression.split(' ') for idx, item in enumerate(expression_list): expression_list[idx] = str(KeyResolver().resolve_key(item, request_body)) final_expression = " ".join(expression_list) result = eval(final_expression) value = set_data(operation.get('stateKey'), result) return value if operation.get('operation') == "conversion": value = KeyResolver().resolve_key(operation.get('value'), request_body) type_cast = operation.get('type') return_value = value.split(',') if type_cast == 'list' else str(value) return set_data(operation.get('stateKey'), return_value) if operation.get('operation') == "apiCall": url = KeyResolver().resolve_key(operation.get('url'), request_body) headers = operation.get('headers') for header, value in headers.items(): resolved_value = KeyResolver().resolve_key(value, request_body) headers[header] = resolved_value body = operation.get('body') method = operation.get('method') if body: for request_key, request_value in body.items(): resolved_value = KeyResolver().resolve_key(request_value, request_body) body[request_key] = resolved_value if method == 'get': response = requests.get(url, headers=headers) else: response = requests.post(url=url, data=body, headers=headers) if response.text is not None: print_log(response.text) html_tag_pattern = re.compile(r'<[^>]+>') contains_html = bool(html_tag_pattern.search(response.text)) if not contains_html: data = json.loads(response.text) set_data(operation.get('stateKey'), data) if operation.get('operation') == 'iterate': store_type = operation.get('storeType') value = operation.get('value') data_value = KeyResolver().resolve_key(value, request_body) struct = operation.get('struct') if store_type == 'list': return_value = [] for datax in data_value: new_updation_dict = {} for i, j in struct.items(): if isinstance(j, int): new_updation_dict.update({i: j}) elif isinstance(j, dict): inner_dict_x = {} for p, q in j.items(): inner_dict = {p: datax.get(q)} inner_dict_x.update(inner_dict) new_updation_dict.update({i: inner_dict_x}) else: dictx = {i: datax.get(j)} new_updation_dict.update(dictx) return_value.append(new_updation_dict) set_data(operation.get('stateKey'), return_value, value_return=True) if operation.get('operation') == 'date_calc': value = operation.get('values') if operation.get('type') == 'subtraction': value_1 = KeyResolver().resolve_key(value[0], request_body) value_1_dob = datetime.strptime(value_1, '%Y-%m-%d') value_2 = KeyResolver().resolve_key(value[1], request_body) if value_2 == 'today': value_2_dob = datetime.now() else: value_2_dob = datetime.strptime(value_2, '%Y-%m-%d') sub_value = value_2_dob.year - value_1_dob.year - ( (value_2_dob.month, value_2_dob.day) < (value_1_dob.month, value_1_dob.day)) set_data(operation.get('stateKey'), sub_value) if operation.get('operation') == 'generateToken': state_key = operation.get('stateKey') payload = operation.get('payload') for k,v in payload.items(): if k in ['exp', 'iat', 'nbf']: time_value = datetime.now() + timedelta(days=v.get('days', 0), hours=v.get('hours', 0), minutes=v.get('minutes', 0), seconds=v.get('seconds', 0)) payload[k] = time_value continue resolved_v = KeyResolver().resolve_key(v, request_body) payload[k] = resolved_v algorithm = operation.get('algorithm') key = KeyResolver().resolve_key(operation.get('signingKey'), request_body) result_token = StaticFunctions.generate_token(payload, algorithm, key) set_data(state_key, result_token) if operation.get('operation') == 'exit': callSkipFunction() except skipItertionException: raise except Exception as ex: print_log(operation)