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

How to use _init_connect method in Airtest

Best Python code snippet using Airtest

read_storage.py

Source:read_storage.py

...21    cursor = None22    db_file_name = None 23    def __init__(self, file_name):24        self.db_file_name = file_name25    def _init_connect(self):26        """Initialize connection to the database,27            optimize the database.28        """29        if self.connection is None:30            try:31                self.connection = sqlite3.connect(self.db_file_name)32                #self.connection = apsw.Connection(self.db_file_name)33                self.cursor = self.connection.cursor()34                self.cursor.execute("PRAGMA synchronous = OFF;")35                self.cursor.execute("PRAGMA journal_mode = OFF;")36                self.cursor.execute("PRAGMA LOCKING_MODE = EXCLUSIVE;")37                self.cursor.execute("PRAGMA foreign_keys=off;")38            except Exception as e:39                raise DatabaseException("Error while connecting to the database.", e)40    def setup(self):41        """Create the initial table for collecting read information 42            from the BAM file.43        """44        if os.path.exists(self.db_file_name):45            os.remove(self.db_file_name)46        self._init_connect()47        try:48            self.cursor.execute("""CREATE TABLE reads ( read_id text NOT NULL, 49                                                        cell_id text NOT NULL,50                                                        sample_name text NOT NULL )51                                                         ;""")52        except Exception as e:53            raise DatabaseException("Error while initializing the database.", e)54        else:55            pass56    def create_indexes(self):57        """Try to speed up subsequent steps by creating indexes in the database."""58        self.cursor.execute("CREATE INDEX idx_reads_read_id ON reads(read_id);")59        #self.cursor.execute("CREATE INDEX idx_cell_id ON reads(cell_id);")60        #self.cursor.execute("CREATE INDEX idx_sample_name ON reads(sample_name);")61        self.cursor.execute("CREATE INDEX idx_reads_cell_sample ON reads(cell_id, sample_name);")62    def process_data(self, threshold):63        """Calculate final association table from the initial read table."""64        self._init_connect()65        print(f"{get_timestamp()}           * Calculating stats on cells")66        self._calculate_stats_on_cells()67        print(f"{get_timestamp()}           * Assigning cells to samples")68        self._assign_cells_to_samples(threshold)69        print(f"{get_timestamp()}           * Creating the association table")70        self._create_final_table()71    def get_total_cell_count(self):72        """Return total number of putative cells."""73        self.cursor.execute("SELECT COUNT(*) as count FROM cells;")74        count = self.cursor.fetchone()75        return(count[0])76    def get_cell_count_per_sample(self):77        """Return cell count per sample tag."""78        self.cursor.execute("SELECT sample, COUNT(*) as count FROM cells GROUP BY sample;")79        counts = self.cursor.fetchall()80        return(counts)81    def _calculate_stats_on_cells(self):82        """Calculate intermitten cells_stat table,83            requires: reads TABLE filled with all values (self.store() )84        """85        try:86            self.cursor.execute("""CREATE TABLE cells_stat AS 87                                SELECT cell_id, sample_name, COUNT(*) abundance 88                                FROM reads GROUP BY cell_id, sample_name;""")89            self.cursor.execute("CREATE INDEX idx_stat_cell_id ON cells_stat(cell_id);")90            self.cursor.execute("CREATE INDEX idx_stat_sample_name ON cells_stat(sample_name);")91        except Exception as e:92            raise DatabaseException("Error while calculating statistics on cell IDs.", e)93    94    def _assign_cells_to_samples(self, threshold = 0.75):95        """Create an intermitten table cells,96            requires: cells_stat TABLE (self.calculate_stats_on_cells() )97        """98        try:99            query = f"""CREATE TABLE cells AS100                          SELECT  cell_id,101                            CASE102                              WHEN max_abd >= sum_abd*{threshold} THEN sample_name103                                  ELSE 'MULTIPLE'104                              END sample FROM105                                (SELECT cell_id, sample_name,106                                     MAX(abundance) max_abd, SUM(abundance) sum_abd107                                         FROM cells_stat GROUP BY cell_id);"""108            self.cursor.execute(query)109            self.cursor.execute("CREATE INDEX idx_cells_cell_id ON cells(cell_id);")110            self.cursor.execute("CREATE INDEX idx_cells_sample_name ON cells(sample);")111        except Exception as e:112            raise DatabaseException("Error while calculating cell-sample relationships.", e)113    def cleanup(self):114        """Delete all but the final table from the database."""115        self._init_connect()116        try:117            self.cursor.execute("DROP TABLE IF EXISTS reads;")118            self.cursor.execute("DROP TABLE IF EXISTS cells;")119            self.cursor.execute("DROP TABLE IF EXISTS cells_stat;")120        except Exception as e:121            raise DatabaseException("Error while deleting the database.", e)122    def _create_final_table(self):123        """Calculate the final association table."""124        self._init_connect()125        try:126            self.cursor.execute("""CREATE TABLE read_sample_pairs AS127                                    SELECT read_id, sample FROM128                                      (SELECT read_id, reads.cell_id, sample FROM reads129                                        LEFT JOIN cells ON reads.cell_id = cells.cell_id);""")130            ## create a covering index131            self.cursor.execute("CREATE INDEX idx_pairs_cover ON read_sample_pairs(read_id, sample);")132        except Exception as e:133            raise DatabaseException("Error while assigning the dominant sample to cell IDs.", e)134    def get_multiple_read_sample_pairs(self, key_list):135        """Return sample for read IDs provided in the argument"""136        self._init_connect()137        return self.get_multiple("read_sample_pairs", ["read_id", "sample"], "read_id", key_list)138    def get_multiple(self, table, field_list, lookup_field, key_list):139        """generic method for retrieving information form the database."""140        if len(key_list) < 1:141            raise DatabaseException("No read IDs provided to look up.")142        fields_string = ",".join([ field for field in field_list ])143        keys_string = ",".join("?" * len(key_list))144        key_list = list(key_list)145        146        result = None147        try:148            self.cursor.execute(f"""SELECT {fields_string} FROM {table} 149                                    WHERE {lookup_field} IN ({keys_string});""", key_list)150            result = dict(self.cursor.fetchall())...

mysql_data_getter.py

Source:mysql_data_getter.py

...20                df.to_csv(file_name)21        finally:22            conn.close()23    @classmethod24    def _init_connect(cls, host, port, uname, pwd, db):25        """26        ç¨æ¥æå»ºmysql connect è¿æ¥å¯¹è±¡27        :param host:28        :param port: ç«¯å£å·ï¼ å¿é¡»æ¯int29        :param uname:30        :param pwd:31        :param db:32        :return: connect å¯¹è±¡33        """34        if isinstance(port, str):35            port = int(port)36        conn = pymysql.connect(host=host, port=port, user=uname,37                               password=pwd, db=db, charset="utf8",38                               cursorclass=pymysql.cursors.DictCursor)39        return conn40    @classmethod41    def download(cls, host, port, uname, pwd, db, sqls):42        conn = cls._init_connect(host=host, port=int(port), uname=uname, pwd=pwd, db=db)...

main_controller.py

Source:main_controller.py

...24        self._init_ui()25    def _init_ui(self):26        # åå§åUIï¼è§å¾ï¼27        self.setupUi(self)28        self._init_connect()29    def _init_connect(self):30        # è°ç¨èªå®ä¹ä¿¡å·åæ§½å½æ°ç±»ï¼æ¨¡åï¼...

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.