How to use _combine_dicts method in autotest

Best Python code snippet using autotest_python

combine_harmonization.py

Source:combine_harmonization.py

...33      joinable_dict[join_key].append(record)34    else:35      joinable_dict[join_key] = [record]36  return joinable_dict37def _combine_dicts(record_dict, new_data_dict, new_data_key):38  record_list = []39  for join_key, record_value in iter(record_dict.items()):40    for pivot in record_value:41      joined_dict = {}42      if join_key in new_data_dict:43        for new_data_value in new_data_dict[join_key]:44          joined_dict = dict(list(pivot.items()) + list(new_data_value.items()))45          record_list.append(joined_dict)46      else:47        if new_data_key == None:48          continue49        joined_dict = dict(list(pivot.items()) + list({new_data_key: []}.items()))50        record_list.append(joined_dict)51  return record_list52def combine(product_file,53            spl_extract,54            rx_extract,55            unii_extract,56            upc_extract,57            json_out):58  json_output_file = open(json_out, 'w')59  product = csv.DictReader(open(product_file), delimiter='\t')60  label = read_json_file(open(spl_extract))61  rxs = read_json_file(open(rx_extract))62  unii = read_json_file(open(unii_extract))63  upcs = read_json_file(open(upc_extract))64  all_products = []65  for row in product:66    clean_product = {}67    #the productid has a date on the front of it, so need to split68    clean_product['id'] = row['PRODUCTID'].split('_')[1]69    clean_product['application_number'] = row['APPLICATIONNUMBER']70    clean_product['product_type'] = row['PRODUCTTYPENAME']71    clean_product['generic_name'] = row['NONPROPRIETARYNAME']72    clean_product['manufacturer_name'] = row['LABELERNAME']73    clean_product['brand_name'] = row['PROPRIETARYNAME']74    clean_product['brand_name_suffix'] = row['PROPRIETARYNAMESUFFIX']75    clean_product['product_ndc'] = row['PRODUCTNDC']76    clean_product['dosage_form'] = row['DOSAGEFORMNAME']77    clean_product['route'] = row['ROUTENAME']78    clean_product['substance_name'] = row['SUBSTANCENAME']79    all_products.append(clean_product)80  joinable_labels = {}81  for spl_data in label:82    clean_label = {}83    clean_label['spl_set_id'] = spl_data['spl_set_id']84    clean_label['id'] = spl_data['id']85    clean_label['spl_version'] = spl_data['spl_version']86    clean_label['is_original_packager'] = spl_data['is_original_packager']87    clean_label['spl_product_ndc'] = spl_data['ProductNDCs']88    clean_label['original_packager_product_ndc'] = \89       spl_data['OriginalPackagerProductNDSs']90    clean_label['package_ndc'] = spl_data['PackageNDCs']91    joinable_labels[clean_label['id']] = [clean_label]92  # Adding labels93  joinable_products = _joinable_dict(all_products, ['id'])94  record_list = _combine_dicts(joinable_labels, joinable_products, None)95  # Adding UNII96  joinable_record_dict = _joinable_dict(record_list, ['id'])97  joinable_unii = _joinable_dict(unii, ['spl_id'])98  record_list = _combine_dicts(joinable_record_dict,99                               joinable_unii,100                               'unii_indexing')101  # Adding RXNorm102  joinable_record_dict = _joinable_dict(record_list,103                                        ['spl_set_id', 'spl_version'])104  joinable_rxnorm = _joinable_dict(rxs, ['spl_set_id', 'spl_version'])105  record_list = _combine_dicts(joinable_record_dict, joinable_rxnorm, 'rxnorm')106  # Adding UPC107  joinable_record_dict = _joinable_dict(record_list, ['spl_set_id'])108  joinable_upc = _joinable_dict(upcs, ['set_id'])109  record_list = _combine_dicts(joinable_record_dict, joinable_upc, 'upc')110  for row in record_list:111    json.dump(row, json_output_file, encoding='utf-8-sig')...

client.py

Source:client.py

...10        valid = self.cf_client.validate_template(11                TemplateBody = template12                )13        return valid14    def _combine_dicts(self, stack):15        combined_dict = {}16        # Add Template Version17        try:18            version = stack.version19        except:20            from fmc.format import Version21            version = Version()22        finally:23            self._update_dict(24                    combined_dict,25                    version.representation()26                    )27        # Add Description28        try:29            self._update_dict(30                    combined_dict,31                    stack.description.representation()32                    )33        except:34            pass35        # Add MetaData36        try:37            self._update_dict(38                    combined_dict,39                    stack.metadata.representation()40                    )41        except:42            pass43        # Add Parameters44        try:45            for parameter in stack.parameters:46                self._update_dict(47                        combined_dict,48                        parameter.representation()49                        )50        except:51            pass52        # Add Mappings53        try:54            for mapping in stack.mappings:55                self._update_dict(56                        combined_dict,57                        mapping.representation()58                        )59        except:60            pass61        # Add Conditions62        try:63            for condition in stack.conditions:64                self._update_dict(65                        combined_dict,66                        condition.representation()67                        )68        except:69            pass70        # Add Resources71        try:72            for resource in stack.resources:73                self._update_dict(74                        combined_dict,75                        resource.representation(),76                        )77        except:78            pass79        # Add Outputs80        try:81            for output in stack.outputs:82                self._update_dict(83                        combined_dict,84                        output.representation()85                        )86        except:87            pass88        return combined_dict89    def validate_stack(self, stack):90        combined_dict = self._combine_dicts(stack)91        serialized_dict = self.serialize(combined_dict)92        valid = self._validate_template(93                serialized_dict94                )95        return valid96    def create_stack(self, stack, validate=True):97        combined_dict = self._combine_dicts(stack)98        Capabilities = []99        for resource in stack.resources:100            try:101                Capabilities.append(resource.Capabilities)102            except:103                pass104        if validate:105            template_valid = self._validate_template(106                    self.serialize(combined_dict)107                    )108        else:109            template_valid = True110        if template_valid:111            response = self.cf_client.create_stack(112                    StackName = stack.name,113                    TemplateBody = self.serialize(combined_dict),114                    Capabilities = Capabilities,115                    )116        return response117    def delete_stack(self, stack):118        return self.cf_client.delete_stack(119                StackName = stack.name120                )121    def stack_representation(self, stack):122        return self._combine_dicts(stack)...

pokeapi_mapper.py

Source:pokeapi_mapper.py

...30def pokemon_mapper(self, exchange):31    req_params = exchange.params32    pid = req_params['pokemon']33    sql_json = sql_util.get_pokemon(pid)34    return _combine_dicts(exchange.json(), sql_json)35@mapper.maps(uri('pokemon-by-id'))36def pokemon_num_mapper(self, exchange):37    req_params = exchange.params38    pid = exchange.json()['name']39    sql_json = sql_util.get_pokemon_by_dexnum(pid)40    return _combine_dicts(exchange.json(), sql_json)41@mapper.maps(uri('move-by-name'))42def move_mapper(self, exchange):43    req_params = exchange.params44    mid = req_params['move_name']45    sql_json = sql_util.get_move(mid)46    return _combine_dicts(exchange.json(), sql_json)47@mapper.maps(uri('ability-by-name'))48def ability_mapper(self, exchange):49    req_params = exchange.params50    aid = req_params['abil_name']51    sql_json = sql_util.get_ability(aid)52    return _combine_dicts(exchange.json(), sql_json)53@mapper.maps(uri('item-by-name'))54def item_mapper(self, exchange):55    req_params = exchange.params56    iid = req_params['item_name']57    sql_json = sql_util.get_item(iid)58    return _combine_dicts(exchange.json(), sql_json)59@mapper.maps(POKEAPI_DEFAULT_ENDPOINTS)60def default_pokeapi_mapper(self, exchange):61    '''62    A method for all endpoints that don't need data from63    the MySQL database64    '''65    return exchange.json()66### Smogon Dex API67print (APIS["smogon_dex"]["base_uri"] + APIS["smogon_dex"]["endpoints"]["set"])  68@mapper.maps(APIS["smogon_dex"]["base_uri"] + APIS["smogon_dex"]["endpoints"]["set"])69def default_smogon_mapper(self, exchange):70    return exchange.json()71def _combine_dicts(dict1, dict2):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.