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

How to use query_fetch_all method in SeleniumBase

Best Python code snippet using SeleniumBase

progress.py

Source:progress.py

...32            column_id = "{0}_exercise_id".format(key)3334            sql_str = "SELECT * FROM {0}_exercise_questions".format(key)35            sql_str += " WHERE {0} = '{1}'".format(column_id, result[column_id])36            questions = self.postgres.query_fetch_all(sql_str)3738            score = 039            for question in questions:40                ans = self.check_answer(question['course_question_id'], question['answer'])4142                if ans['isCorrect'] is True:43                    score += 14445            # UPDATE PROGRESS46            progress = self.check_progress(questions)4748            # PERCENT SCORE49            total = len(questions)50            percent_score = self.format_progress((score / total) * 100)5152            # UPDATE SCORE53            tmp = {}54            tmp['score'] = score55            tmp['progress'] = progress56            tmp['percent_score'] = percent_score57            if key.upper() == "STUDENT":58                tmp['end_on'] = time.time()5960            tmp['update_on'] = time.time()61            conditions = []6263            conditions.append({64                "col": "exercise_id",65                "con": "=",66                "val": exercise_id67            })6869            conditions.append({70                "col": column_id,71                "con": "=",72                "val": result[column_id]73            })7475            data = self.remove_key(tmp, "exercise_id")7677            self.postgres.update(exercise_table, data, conditions)7879    def update_subsection_progress(self, user_id, exercise_id, key):80        """ Update Course Progress """8182        exercise_table = "{0}_exercise".format(key)83        subsection_table = "{0}_subsection".format(key)84        exercise_questions_table = "{0}_exercise_questions".format(key)85        key_subsection_id = "{0}_subsection_id".format(key)86        key_exercise_id = "{0}_exercise_id".format(key)8788        sql_str = "SELECT * FROM exercise e LEFT JOIN {0}".format(subsection_table)89        sql_str += " sub ON e.subsection_id = sub.subsection_id"90        sql_str += " WHERE e.exercise_id='{0}'".format(exercise_id)91        subsection = self.postgres.query_fetch_one(sql_str)9293        student_subsection_id = subsection[key_subsection_id]94        subsection_id = subsection['subsection_id']95        course_id = subsection['course_id']9697        sql_str = "SELECT answer, percent_score FROM {0}".format(exercise_questions_table)98        sql_str += " WHERE {0} IN (SELECT {0}".format(key_exercise_id)99        sql_str += " FROM {0} WHERE status is True AND".format(exercise_table)100        sql_str += " account_id='{0}' AND".format(user_id)101        sql_str += "  exercise_id IN (SELECT exercise_id FROM exercise WHERE "102        sql_str += "  subsection_id='{0}' AND status is True))".format(subsection_id)103        student_exercise = self.postgres.query_fetch_all(sql_str)104105        answer = [dta['answer'] for dta in student_exercise]106        total = len(answer)107        not_answer = [None, '', []]108        total_answer = len([ans for ans in answer if ans not in not_answer])109        # print("Total: {0} Answer: {1}".format(total, total_answer))110        progress = 0111        if total_answer:112            progress = self.format_progress((total_answer / total) * 100)113114        # PERCENT SCORE115        percent_score = [dta['percent_score'] for dta in student_exercise]116        total = len(percent_score)117        percent_100 = len([dta['percent_score'] for dta in student_exercise \118                           if dta['percent_score'] is not None and \119                           int(dta['percent_score']) == 100])120121        percent_score = self.format_progress((percent_100 / total) * 100)122123        # UPDATE PROGRESS124        conditions = []125126        conditions.append({127            "col": key_subsection_id,128            "con": "=",129            "val": student_subsection_id130        })131132        conditions.append({133            "col": "course_id",134            "con": "=",135            "val": course_id136        })137138        data = {}139        data['progress'] = progress140        data['percent_score'] = percent_score141        data['update_on'] = time.time()142143        self.postgres.update(subsection_table, data, conditions)144145    def update_section_progress(self, user_id, exercise_id, key):146        """ Update Course Progress """147148        section_table = "{0}_section".format(key)149        key_section_id = "{0}_section_id".format(key)150        exercise_question_table = "{0}_exercise_questions".format(key)151        key_exercise_id = "{0}_exercise_id".format(key)152        exercise_table = "{0}_exercise".format(key)153154        sql_str = "SELECT * FROM exercise e"155        sql_str += " LEFT JOIN {0} ss ON e.section_id = ss.section_id".format(section_table)156        sql_str += " WHERE e.exercise_id='{0}'".format(exercise_id)157        section = self.postgres.query_fetch_one(sql_str)158159        section_id = section['section_id']160        course_id = section['course_id']161        student_section_id = section[key_section_id]162163        sql_str = "SELECT answer, percent_score FROM {0}".format(exercise_question_table)164        sql_str += " WHERE {0} IN (SELECT {0}".format(key_exercise_id)165        sql_str += " FROM {0} WHERE status is True AND".format(exercise_table)166        sql_str += " account_id='{0}' AND".format(user_id)167        sql_str += " exercise_id IN (SELECT exercise_id FROM exercise WHERE "168        sql_str += " section_id='{0}' AND status is True))".format(section_id)169        student_exercise = self.postgres.query_fetch_all(sql_str)170171        answer = [dta['answer'] for dta in student_exercise]172        total = len(answer)173        not_answer = [None, '', []]174        total_answer = len([ans for ans in answer if ans not in not_answer])175176        progress = 0177        if total_answer:178            progress = self.format_progress((total_answer / total) * 100)179180        # PERCENT SCORE181        percent_100 = len([dta['percent_score'] for dta in student_exercise \182                           if dta['percent_score'] is not None and \183                           int(dta['percent_score']) == 100])184185        percent_score = self.format_progress((percent_100 / total) * 100)186187        # UPDATE PROGRESS188        conditions = []189190        conditions.append({191            "col": key_section_id,192            "con": "=",193            "val": student_section_id194        })195196        conditions.append({197            "col": "course_id",198            "con": "=",199            "val": course_id200        })201202        data = {}203        data['progress'] = progress204        data['percent_score'] = percent_score205        data['update_on'] = time.time()206207        self.postgres.update(section_table, data, conditions)208209210    def update_course_progress(self, user_id, exercise_id, key):211        """ Update Course Progress """212213        exercise_table = "{0}_exercise".format(key)214        exercise_question_table = "{0}_exercise_questions".format(key)215        key_exercise_id = "{0}_exercise_id".format(key)216        course_table = "{0}_course".format(key)217218        sql_str = "SELECT * FROM {0} WHERE".format(exercise_table)219        sql_str += " exercise_id='{0}' AND status=true".format(exercise_id)220        sql_str += " AND account_id='{0}'".format(user_id)221        student_course = self.postgres.query_fetch_one(sql_str)222        course_id = student_course['course_id']223224        sql_str = "SELECT answer, percent_score FROM {0}".format(exercise_question_table)225        sql_str += " WHERE {0} IN (SELECT {0}".format(key_exercise_id)226        sql_str += " FROM {0} WHERE status is True AND".format(exercise_table)227        sql_str += "  account_id='{0}' AND status is True)".format(user_id)228        student_exercise = self.postgres.query_fetch_all(sql_str)229230        answer = [dta['answer'] for dta in student_exercise]231        total = len(answer)232233        not_answer = [None, '', []]234        total_answer = len([ans for ans in answer if ans not in not_answer])235236        progress = self.format_progress((total_answer / total) * 100)237238        # PERCENT SCORE239        percent_score = [dta['percent_score'] for dta in student_exercise]240        total = len(percent_score)241        percent_100 = len([dta['percent_score'] for dta in student_exercise \242                           if dta['percent_score'] is not None and \243                           int(dta['percent_score']) == 100])244245        percent_score = self.format_progress((percent_100 / total) * 100)246247        # UPDATE PROGRESS248        conditions = []249250        conditions.append({251            "col": "account_id",252            "con": "=",253            "val": user_id254        })255256        conditions.append({257            "col": "course_id",258            "con": "=",259            "val": course_id260        })261262        data = {}263        data['progress'] = progress264        data['percent_score'] = percent_score265        data['update_on'] = time.time()266267        self.postgres.update(course_table, data, conditions)268269        if key == 'student':270271            self.all_progress(user_id)272            self.group_progress(user_id)273274    def all_progress(self, user_id):275        """ UPDATE ALL PROGRESS """276277        sql_str = "SELECT progress FROM student_course WHERE"278        sql_str += " account_id='{0}' AND status='t'".format(user_id)279        progress = self.postgres.query_fetch_all(sql_str)280281        total_prog = 0282        total_course = len(progress) * 100283        for pgrss in progress:284285            total_prog = total_prog + float(pgrss['progress'])286287        average = self.format_progress((total_prog/total_course) * 100)288289        # UPDATE PROGRESS290        conditions = []291292        conditions.append({293            "col": "id",294            "con": "=",295            "val": user_id296        })297298        data = {}299        data['progress'] = average300        data['update_on'] = time.time()301302        self.postgres.update('account', data, conditions)303304        return 1305306    def group_progress(self, user_id):307        """ UPDATE GROUP PROGRESS """308309        sql_str = "SELECT user_group_id FROM user_group_students WHERE"310        sql_str += " student_id='{0}'".format(user_id)311312        user_group = self.postgres.query_fetch_all(sql_str)313314        # UPDATE GROUP315        for sgroup in user_group:316317            user_group_id = sgroup['user_group_id']318319            # GET ALL USER PROGRESS IN A GROUP320            sql_str = "SELECT progress FROM account WHERE id IN ("321            sql_str += "SELECT student_id FROM user_group_students WHERE"322            sql_str += " user_group_id='{0}')".format(user_group_id)323            progress = self.postgres.query_fetch_all(sql_str)324325            total_prog = 0326            total_student = len(progress) * 100327328            for pgrss in progress:329330                total_prog = total_prog + float(pgrss['progress'])331332            # GET AVERAGE PROGRESS333            average = self.format_progress((total_prog/total_student) * 100)334335            # UPDATE PROGRESS336            conditions = []337
...

df_to_rds_aws.py

Source:df_to_rds_aws.py

1import pymysql2from auth import *3import pandas as pd4import time5#read df already saved from running previous script6df = pd.read_csv('youtube.csv', index_col=0)7def connect_to_db():8    try:9        connection = pymysql.connect(host=ENDPOINT,user=USERNAME,\10            password=DB_PASSWORD,database=DATABASE_NAME,port=PORT)11    except pymysql.OperationalError as e:12        raise e13    else:14        cursor = connection.cursor()15        print('Connected!')16    return connection, cursor17def create_table(cursor):18    query_create_table = ("""CREATE TABLE IF NOT EXISTS videos_details19    (video_id VARCHAR(100) PRIMARY KEY,20    video_title TEXT NOT NULL ,21    upload_date DATE NOT NULL,22    view_count int NOT NULL,23    like_count int NOT NULL,24    dislike_count int NOT NULL,25    comment_count int NOT NULL 26    )27    """)28    cursor.execute(query_create_table)29def insert_row(cursor, video_id, video_title, upload_date, view_count, like_count, dislike_count, comment_count):30    query_insert_row = ("""31    INSERT INTO videos_details (video_id,video_title,upload_date,view_count32    , like_count, dislike_count,comment_count)33    VALUES (%s,%s,%s,%s,%s,%s,%s);34    """)35    values = (video_id, video_title, upload_date, view_count, like_count, dislike_count, comment_count)36    cursor.execute(query_insert_row,values)37def check_vid_not_exist(cursor,video_id):38    query_search_vid_id = ("""39    SELECT * FROM videos_details40    WHERE video_id = %s41    """)42    cursor.execute(query_search_vid_id,(video_id,))43    result = cursor.fetchone() is None44    return result45def insert_in_table(cursor):46    df2=pd.DataFrame(columns=list(df.columns))47    #iterate over all videos48    for i,row in df.iterrows():49        #for each video if it video_id doesn't exist,50        #insert row in table51        if check_vid_not_exist(cursor, row['video_id']):52            insert_row(cursor, row['video_id'], row['video_title'], row['upload_date'], row['view_count']53                          , row['like_count'], row['dislike_count'], row['comment_count'])54        else:55            df2.append(row)56    return df257def update_row(cursor, video_id, video_title, view_count, like_count, dislike_count, comment_count):58    query_update_row = ("""UPDATE videos_details59            SET video_title = %s,60                view_count = %s,61                like_count = %s,62                dislike_count = %s,63                comment_count = %s64            WHERE video_id = %s;""")65    values = (video_title, view_count, like_count, dislike_count, comment_count, video_id)66    cursor.execute(query_update_row, values)67def update_table(cursor, df):68    for i, row in df.iterrows():69        update_row(row['video_id'], row['video_title'], row['upload_date'], row['view_count']70                          , row['like_count'], row['dislike_count'], row['comment_count'])71    72#connect to db73connection, cursor = connect_to_db()74#create table in db75create_table(cursor)76#insert values one by one into the table:77#1/if they don't exist use INSERT INTO78df2 = insert_in_table(cursor)79connection.commit()80#2/if they exist update video details81update_table(cursor, df2)82connection.commit()83#Fetch all to make sure everything worked84query_fetch_all = ("""85SELECT * FROM videos_details;86 """)87cursor.execute(query_fetch_all)88cursor.fetchall()89#DROP table because of aws charges90query_fetch_all = ("""91DROP TABLE videos_details;92 """)93cursor.execute(query_fetch_all)94connection.commit()95#close the connection...

dbinteraction.py

Source:dbinteraction.py

...10        # You need to write the code that will create a connection to DB11        # self.connection = db2.connect(constants.DBConnect.SERVER)12        self.cursor = self.connection.cursor()13        logging.info('-----------Successfully Connected to DB----------')14    def query_fetch_all(self, query):15        """ Executes a DB query, gets all the values """16        self.cursor.execute(query)17        records = self.cursor.fetchall()18        return records19    def query_fetch_one(self, query):20        """ Executes a db query, gets first value """21        self.cursor.execute(query)22        record = self.cursor.fetchone()23        return record[0]24    def execute_query(self,query):25        """ Executes a query in the db """26        self.cursor.execute(query)27    def generate_dynamic_query(self, tableName, columnValueA, param):28        query = f'SELECT {param} FROM {tableName} WHERE RUN_ID = {columnValueA}'...

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.