How to use _local_execute method in fMBT

Best Python code snippet using fMBT_python

task_executor.py

Source:task_executor.py

...17        self.local_schema = None18        self.global_schema = None19        self.node_id = node_id20        self.iternum = 021    async def _local_execute(self, local, sqlscript, insert):22        if not insert:23            query = (24                "delete from "25                + self.localtable26                + "_"27                + str(self.node_id)28                + "; insert into "29                + self.localtable30                + "_"31                + str(self.node_id)32                + " select "+ str(self.node_id) +" as node_id, * from ("33                + sqlscript34                + ") inputquery;"35            )36        else:37            query = (38                    "insert into "39                    + self.localtable40                    + "_"41                    + str(self.node_id)42                    + " select "+str(self.node_id)+" as node_id, * from ("43                    + sqlscript44                    + ") inputquery;"45            )46        await local.cursor().execute(query)47    async def _create_view(self, local, attributes, table, filterpart, vals):48        if filterpart == " ":49            query = (50                "CREATE VIEW "51                + self.viewlocaltable52                + " AS select "53                + ",".join(attributes)54                + " from "55                + table56                + ";"57            )58            await local.cursor().execute(query)59        else:60            query = (61                "CREATE VIEW "62                + self.viewlocaltable63                + " AS select "64                + ",".join(attributes)65                + " from "66                + table67                + " where "68                + filterpart69                + ";"70            )71            await local.cursor().execute(query, vals)72    async def _initialize_local_schema(self):73        query = "drop table if exists %s; create table %s (node_id INT, %s);" % (74            self.localtable + "_" + str(self.node_id),75            self.localtable + "_" + str(self.node_id),76            self.local_schema,77        )78        await self.db_objects["local"]["async_con"].cursor().execute(query)79    def create_task_generator(self, algorithm):80        return algorithm(81            self.viewlocaltable,82            self.globaltable,83            self.parameters,84            self.attributes,85            self.globalresulttable86        )87    async def init_global_remote_table(self, global_schema):88        await self.transfer_runner.initialize_local(global_schema)89    async def init_tables(self, local_schema, global_schema):90        self.local_schema = local_schema91        await self._initialize_local_schema()92        await self.transfer_runner.initialize_local(global_schema)93    # parameters binding is an important processing step on the parameters that will be concatenated in an SQL query94    # to avoid SQL injection vulnerabilities. This step is not implemented for postgres yet but only for monetdb95    # so algorithms that contain parameters (other than attribute names) will raise an exception if running with postgres96    def bindparameters(self, parameters):97        boundparam = []98        for i in parameters:99            if isinstance(i, (int, float, complex)):100                boundparam.append(i)101            else:102                boudparam.append(self.db_objects["global"]["async_con"].bind_str(i))103        return boundparam104    async def createlocalviews(self):105        t1 = current_time()106        table = self.params["table"]107        attributes = []108        for attribute in self.params["attributes"]:109            attributes.append(attribute)110        for formula in self.params["filters"]:111            for attribute in formula:112                if attribute[0] not in attributes:113                    attributes.append(attribute[0])114        await self.db_objects["local"]["async_con"].check_for_params(table, attributes)115        filterpart = " "116        vals = []117        for j, formula in enumerate(self.params["filters"]):118            andpart = " "119            for i, filt in enumerate(formula):120                if filt[1] not in [">", "<", "<>", ">=", "<=", "="]:121                    raise Exception("Operator " + filt[1] + " not valid")122                andpart += filt[0] + filt[1] + "%s"123                vals.append(filt[2])124                if i < len(formula) - 1:125                    andpart += " and "126            if andpart != " ":127                filterpart += "(" + andpart + ")"128            if j < len(self.params["filters"]) - 1:129                filterpart += " or "130        await self._create_view(131                self.db_objects["local"]["async_con"],132                self.params["attributes"],133                self.params["table"],134                filterpart,135                vals,136            )137        print("time " + str(current_time() - t1))138    #### run a task on all local nodes and sets up the transfer of the results to global node139    async def task_local(self, schema, static_schema, sqlscript):140        t1 = current_time()141       # insert = False142       # if 'iternum'  in  schema:143       #     insert = True144       # if self.local_schema == None or self.local_schema != schema:145       #     self.local_schema = schema146       #     await self._initialize_local_schema()147       #     await self.transfer_runner.initialize_local(self.local_schema)148        if int(static_schema):149            await self._local_execute(self.db_objects["local"]["async_con"], sqlscript, 0)150        else:151            self.local_schema = schema152            await self._initialize_local_schema()153            await self._local_execute(self.db_objects["local"]["async_con"], sqlscript, 0)154        ## for debug, print local contents155        #for id, local in enumerate(self.db_objects["local"]):156        #    result = await local["async_con"].cursor().execute("select * from %s_%s;" % (self.localtable, str(id)))157        print("time " + str(current_time() - t1))158    async def get_global_result(self):159        cur = self.db_objects["local"]["async_con"].cursor()160        query = f'select * from {self.globalresulttable};'161        await cur.execute(query)162        return await cur.fetchall()163    async def register_udf(self, udf):164        # this is not implemented165        await self.db_objects["local"]["async_con"].create_function(udf)166    async def clean_up(self):167        await self.db_objects["local"]["async_con"].clean_tables(self.db_objects, self.globaltable, self.localtable+"_"+str(self.node_id), self.viewlocaltable+"_"+str(self.node_id), self.globalresulttable)

transactional.py

Source:transactional.py

...82                                  action._topic,83                                  action._tag)84        prepared = AsyncEventValue()85        def _send(producer):86            def _local_execute(msg, user_args):87                run_coroutine_threadsafe(prepared.set(SendStatus.OK), loop)88                # return TransactionStatus.UNKNOWN89                return self._transaction_status.wait()90            try:91                ret = producer.send_message_in_transaction(92                    msg_obj, _local_execute, None)93                if ret.status != SendStatus.OK:94                    run_coroutine_threadsafe(prepared.set(ret.status), loop)95            except Exception as exc:96                run_coroutine_threadsafe(prepared.set(exc), loop)97        producer_executor.submit(_send, action.get_producer())98        send_status = await prepared.wait()99        if send_status == SendStatus.OK:100            return...

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.