How to use _init_connect method in Airtest

Best Python code snippet using Airtest

read_storage.py

Source: read_storage.py Github

copy

Full Screen

...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())...

Full Screen

Full Screen

mysql_data_getter.py

Source: mysql_data_getter.py Github

copy

Full Screen

...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)...

Full Screen

Full Screen

main_controller.py

Source: main_controller.py Github

copy

Full Screen

...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 # 调用自定义信号和槽函数类(模型)...

Full Screen

Full Screen

Blogs

Check out the latest blogs from LambdaTest on this topic:

QA&#8217;s and Unit Testing &#8211; Can QA Create Effective Unit Tests

Unit testing is typically software testing within the developer domain. As the QA role expands in DevOps, QAOps, DesignOps, or within an Agile team, QA testers often find themselves creating unit tests. QA testers may create unit tests within the code using a specified unit testing tool, or independently using a variety of methods.

LIVE With Automation Testing For OTT Streaming Devices ????

People love to watch, read and interact with quality content — especially video content. Whether it is sports, news, TV shows, or videos captured on smartphones, people crave digital content. The emergence of OTT platforms has already shaped the way people consume content. Viewers can now enjoy their favorite shows whenever they want rather than at pre-set times. Thus, the OTT platform’s concept of viewing anything, anytime, anywhere has hit the right chord.

Why Agile Is Great for Your Business

Agile project management is a great alternative to traditional methods, to address the customer’s needs and the delivery of business value from the beginning of the project. This blog describes the main benefits of Agile for both the customer and the business.

Options for Manual Test Case Development &#038; Management

The purpose of developing test cases is to ensure the application functions as expected for the customer. Test cases provide basic application documentation for every function, feature, and integrated connection. Test case development often detects defects in the design or missing requirements early in the development process. Additionally, well-written test cases provide internal documentation for all application processing. Test case development is an important part of determining software quality and keeping defects away from customers.

Test strategy and how to communicate it

I routinely come across test strategy documents when working with customers. They are lengthy—100 pages or more—and packed with monotonous text that is routinely reused from one project to another. Yawn once more— the test halt and resume circumstances, the defect management procedure, entrance and exit criteria, unnecessary generic risks, and in fact, one often-used model replicates the requirements of textbook testing, from stress to systems integration.

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run Airtest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful