KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _process_records method in localstack

Best Python code snippet using localstack_python

juchao_info.py

Source:juchao_info.py

...133        datas_items = self.process_records(datas_records, 1127)134        for items in (zuixin_items, stock_items, fund_items, datas_items):135            save_num = self._batch_save(self.spider_client, items, self.table_name, self.fields)136    # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -137    def _process_records(self, records, type_code):138        for record in records:139            item = dict()140            pub_date = record.get("DECLAREDATE")141            if not pub_date:142                pub_date = record.get("RECTIME")143            item['PubDatetime'] = pub_date  # åå¸æ¶é´144            item['SecuCode'] = record.get("SECCODE")  # è¯å¸ä»£ç 145            item['Title'] = record.get("F001V")  # èµè®¯æ é¢146            item['InnerType'] = record.get("F003V")   # èµè®¯ç±»å«147            item['Content'] = record.get("F002V")   # èµè®¯æè¦148            # å¢å åå¹¶è¡¨åæ®µ149            item['DupField'] = "{}_{}_{}".format(self.table_code, item['SecuCode'], item['Title'])150            item['MedName'] = self.name151            item['OrgMedName'] = self.name152            item['OrgTableCode'] = self.table_code153            self._save(self.spider_client, item, self.merge_table, self.merge_fields)154    def run(self):155        self._spider_init()156        zuixin_records = self.get_list(self.zuixin_url)157        self._process_records(zuixin_records, 1128)158        stock_records = self.get_list(self.stock_url)159        self._process_records(stock_records, 1078)160        fund_records = self.get_list(self.fund_url)161        self._process_records(fund_records, 1126)162        datas_records = self.get_list(self.datas_url)163        self._process_records(datas_records, 1127)164    def trans_history(self):165        self._spider_init()166        for i in range(1000):    # TODO167            trans_sql = '''select pub_date as PubDatetime,\168code as SecuCode, \169title as Title,\170category as InnerType,\171summary as Content, \172CREATETIMEJZ as CreateTime, \173UPDATETIMEJZ as UpdateTime \174from {} limit {}, 1000; '''.format(self.table_name, i*1000)175            datas = self.spider_client.select_all(trans_sql)176            print(len(datas))177            if not datas:...

kinesis_records.py

Source:kinesis_records.py

...38            data_in_bytes = base64.b64decode(data)39            data_decoded = data_in_bytes.decode("utf-8")40            data_dict = json.loads(data_decoded)41            self._data.append(data_dict)42        return self._process_records()43    def _process_records(self):44        result = False45        try:46            mindsphere_event_connector = MindSphereEventConnector()47            mindsphere_event_connector.getHeaders()48            events_counter = 049            for record in self._data:50                if not "uniqueDeviceId" in record:51                    print(52                        "Error: kinesis_records: _process_records: uniqueDeviceId is not found. skipping record...")53                    continue54                beacon_id = record["uniqueDeviceId"]55                # if not "beaconName" in record:56                #     print(57                #         "Error: kinesis_records: _process_records: beaconName is not found. skipping record...")...

BingoElastic.py

Source:BingoElastic.py

...60            indigo_session=self.indigo,61            limit=5000,62            options=options,63        )64        return self._process_records(records)65    @catch_indigo_exception(catch_error=True)66    def substructure(67        self, molecule: IndigoObject, target_function: str, options: str = ""68    ):69        query_mol = self.indigo.loadQueryMolecule(molecule.rawData())70        query_mol.aromatize()71        records = self.repo.filter(72            query_subject=query_mol,73            indigo_session=query_mol._session,74            limit=5000,75            options=options,76        )77        return self._process_records(records)78    @catch_indigo_exception(catch_error=True)79    def similarity(80        self, molecule: IndigoObject, target_function: str, sim_type, options: str81    ):82        min_sim, max_sim = options.split(", ")83        min_sim, max_sim = float(min_sim), float(max_sim)84        if sim_type.startswith("euclid"):85            sim_type = "euclid"86        compound = self.indigo.loadMolecule(molecule.rawData())87        indigo_record = IndigoRecord(indigo_object=compound)88        alg = getattr(SimilarityMatch, sim_type)(indigo_record, min_sim)89        records = self.repo.filter(query_subject=alg, limit=5000)90        return self._process_records(records)91    def _process_records(self, records):92        result = []93        for record in records:94            if record:95                result.append(record.as_dict()["index"])96        return result97    def drop(self):98        logger.info(f"Dropping {self.dbms} repository")...

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.