How to use query_all method in robotframework-ioslibrary

Best Python code snippet using robotframework-ioslibrary_python

data_functions.py

Source:data_functions.py Github

copy

Full Screen

2import models3from pathlib import Path4import json5def write_asset_types_to_json():6 asset_types = db_connect.query_all(models.AssetTypes)7 type_dict = {}8 for i in asset_types:9 type_dict[i.id] = i.asset_type10 data_folder = Path("static/data/all_assoc_types.json")11 with open(data_folder, 'w') as fp:12 json.dump(type_dict, fp, indent=4)13 return type_dict14def get_associated_solutions(solution_id):15 s = db_connect.query_one_db(model=models.Solutions, column=models.Solutions.id, v=solution_id)16 assoc_solutions = {}17 if s.associated_solutions:18 for r in s.associated_solutions:19 asid = db_connect.query_one_db(model=models.Solutions, column=models.Solutions.id, v=r)20 assoc_solutions[asid.id] = asid.solution_title21 else:22 assoc_solutions = []23 return assoc_solutions24def one_solution_asset_types(solution_id):25 # This needs fixed. This should really call a function that just returns a dictionary of all asset types without writing json again26 all_asset_types = write_asset_types_to_json()27 solution_asset_types = {}28 associated_solutions = {}29 solution = db_connect.query_one_db(model=models.Solutions, column=models.Solutions.id, v=solution_id)30 for t in solution.associated_asset_types:31 a_type = db_connect.query_one_db(model=models.AssetTypes, column=models.AssetTypes.id, v=t)32 solution_asset_types[t] = a_type.asset_type33 if solution.associated_solutions:34 for a in solution.associated_solutions:35 asid = db_connect.query_one_db(model=models.Solutions, column=models.Solutions.id, v=a)36 associated_solutions[a] = asid.solution_title37 all_solution_data = {38 'all_asset_types': all_asset_types,39 'solution_asset_types': solution_asset_types,40 'associated_solutions': associated_solutions,41 }42 data_folder = Path("static/data/one_solution_data.json")43 with open(data_folder, 'w') as fp:44 json.dump(all_solution_data, fp, indent=4)45 return solution_asset_types46def one_solution_data(solution_id):47 solution_data = {}48 solution = db_connect.query_one_db(model=models.Solutions, column=models.Solutions.id, v=solution_id)49 solution_data["Solution_id"] = solution.id50 solution_data["Title"] = solution.solution_title51 solution_data["Steps"] = solution.steps52 step_number = []53 for s in solution.steps:54 step_number.append(s)55 if len(step_number) > 0:56 solution_data["Highest_Step"] = max(step_number)57 else:58 solution_data["Highest_Step"] = '0'59 solution_data["Associated_Asset_Types"] = solution.associated_asset_types60 solution_data["Primary_Asset_Type"] = solution.primary_asset_type61 data_folder = Path("static/data/one_solution.json")62 with open(data_folder, 'w') as fp:63 json.dump(solution_data, fp, indent=4)64 return None65def write_all_solution_data():66 all_solutions = db_connect.query_all(models.Solutions)67 solution_data = []68 for s in all_solutions:69 main_asset_type = db_connect.query_one_db(model=models.AssetTypes, column=models.AssetTypes.id, v=s.primary_asset_type)70 solution_data.append(71 {72 'id': s.id,73 'title': s.solution_title,74 'primary_asset_type': main_asset_type.asset_type75 }76 )77 data_folder = Path("static/data/all_solution_data.json")78 with open(data_folder, 'w') as fp:79 json.dump(solution_data, fp, indent=4)80 return None81def write_one_asset_data_to_json(asset_id):82 asset_data = {}83 asset = db_connect.query_one_db(model=models.Assets, column=models.Assets.id, v=asset_id)84 asset_data["Asset_Id"] = asset.id85 asset_data["Asset_Type"] = asset.asset_type86 asset_data["Manufacturer"] = asset.manufacturer87 asset_data["Model"] = asset.model88 asset_data["Serial_No"] = asset.serial_no89 asset_data["Dia_Asset_Tag"] = asset.dia_asset_tag90 asset_data["Description"] = asset.description91 asset_data["Department"] = asset.department92 asset_data["Location"] = asset.location93 # data_folder = Path("static/data/one_asset.json")94 # with open(data_folder, 'w') as fp:95 # json.dump(asset_data, fp, indent=4)96 return asset_data97def write_asset_data_to_json(asset_id=0):98 if asset_id != 0:99 asset_data = {}100 all_types = []101 all_manufacturers = []102 all_departments = []103 for t in db_connect.query_all(models.AssetTypes):104 all_types.append(105 {106 'id': t.id,107 'asset_type': t.asset_type108 }109 )110 for m in db_connect.query_all(models.Manufacturers):111 all_manufacturers.append(112 {113 'id': m.id,114 'manufacturer': m.manufacturer115 }116 )117 for d in db_connect.query_all(models.Departments):118 all_departments.append(119 {120 'id': d.id,121 'department': d.department122 }123 )124 asset_data['all_types'] = all_types125 asset_data['all_manufacturers'] = all_manufacturers126 asset_data['all_departments'] = all_departments127 asset_data['this_asset'] = write_one_asset_data_to_json(asset_id)128 data_folder = Path("static/data/all_asset_data.json")129 with open(data_folder, 'w') as fp:130 json.dump(asset_data, fp, indent=4)131 return None132 else:133 return None134# This makes a dictionary of { Manufacturer: { Asset_Type: [ Model no. list ] }, Manufacturer: { Asset_Type: [ Model no. list ] }}135def asset_models_by_manufacturer():136 all_manufacturers = db_connect.query_all(models.Manufacturers)137 all_asset_types = db_connect.query_all(models.AssetTypes)138 model_by_manufacturer = {}139 for m in all_manufacturers:140 asset_type_list = {}141 for t in all_asset_types:142 model_numbers_for_type = []143 q = db_connect.query_distinct_for_models(m.id, t.id)144 for model_no in q:145 model_numbers_for_type.append(model_no.model)146 asset_type_list[t.id] = model_numbers_for_type147 model_by_manufacturer[m.id] = asset_type_list148 data_folder = Path("static/data/model_by_manufacturer.json")149 with open(data_folder, 'w') as fp:150 json.dump(model_by_manufacturer, fp, indent=4)151 return model_by_manufacturer152def manufacturers_as_dict():153 all_manufacturers = db_connect.query_all(models.Manufacturers)154 manufacturer_dict = {}155 for m in all_manufacturers:156 manufacturer_dict[m.id] = m.manufacturer157 return manufacturer_dict158def departments_as_dict():159 all_departments = db_connect.query_all(models.Departments)160 dept_dict = {}161 for d in all_departments:162 dept_dict[d.id] = d.department163 return dept_dict164def software_companies_as_dict():165 all_software_companies = db_connect.query_all(models.SoftwareCompanies)166 software_co_dict = {}167 for s in all_software_companies:168 software_co_dict[s.id] = s.software_company169 return software_co_dict170def software_names_as_dict():171 all_software = db_connect.query_all(models.Software)172 software_name_dict = {}173 for s in all_software:174 software_name_dict[s.id] = s.software_name...

Full Screen

Full Screen

04_mysql_base.py

Source:04_mysql_base.py Github

copy

Full Screen

1# coding=utf-82import mysql.connector34mydb = mysql.connector.connect(5 host="192.168.249.143",6 user="root",7 passwd="3282",8 database="python4"9)1011if __name__ == "__main__":12 print(mydb)13 # 1. 创建数据库14 mycorsor = mydb.cursor()15 # mycorsor.execute("create database python4")1617 # 2. 创建表18 mycorsor.execute("use python4")19 # mycorsor.execute("create table person(name varchar(40),age int)")20 mycorsor.execute("CREATE TABLE sites (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255), url VARCHAR(255))")21 # mycorsor.execute("CREATE TABLE sites (name VARCHAR(255), url VARCHAR(255))")2223 # 3. 输出所有数据库列表24 print("=================== 3. 输出所有数据库列表 ===================")25 mycorsor.execute("SHOW DATABASES")26 for database in mycorsor:27 print(database)2829 # 4. 查看数据表是否已存在30 print("=================== 4. 查看数据表是否已存在 ===================")31 mycorsor.execute("SHOW TABLES")32 for table in mycorsor:33 print("tpye(table) = {} table = {}".format(type(table), table))3435 # 5. 主键设置36 # mycorsor.execute("ALTER TABLE sites ADD COLUMN id INT AUTO_INCREMENT PRIMARY KEY")3738 # 6. 插入数据39 sql = "INSERT INTO sites (name,url) VALUES (%s,%s)"40 values = ("baidu", "www.baidu.com")41 mycorsor.execute(sql, values)42 mydb.commit() # 提交事务43 insert_id = mycorsor.lastrowid44 print("insert into data success")45 print("1条数据插入成功,ID:{}".format(insert_id))4647 # 7. 批量插入48 sql = "INSERT INTO sites (name,url) VALUES (%s,%s)"49 values = [("Google", "www.google.com"),50 ("github", "www.github.com"),51 ("Tencent", "www.qq.com"),52 ("Apple", "www.apple.com")]53 mycorsor.executemany(sql, values)54 mydb.commit()55 count = mycorsor.rowcount56 print("插入{}数据成功...".format(count))5758 # 8. 查询数据59 print("=================== 8. 查询数据 ===================")60 mycorsor.execute("SELECT * FROM sites")61 # 获取所有记录62 qyery_all = mycorsor.fetchall()63 for row in qyery_all:64 print(row)6566 print("=================== 9. 也可以读取指定的字段数据 ===================")67 mycorsor.execute("SELECT name, url FROM sites")68 query_all = mycorsor.fetchall()69 for row in query_all:70 print(row)7172 print("=================== 10. 如果我们只想读取一条数据，可以使用 fetchone() ===================")73 # mycorsor.execute("SELECT name,url FROM sites")74 # row = mycorsor.fetchone()75 # print(row)7677 print("=================== 11. where 条件语句 ===================")78 mycorsor.execute("SELECT * FROM sites WHERE name='Google'")79 query_all = mycorsor.fetchall()80 for row in query_all:81 print(row)8283 print("=================== 12. where 条件语句 ,也可以使用通配符 % ===================")84 sql = "SELECT * FROM sites WHERE url LIKE '%oo%'"85 mycorsor.execute(sql)86 query_all = mycorsor.fetchall()87 for row in query_all:88 print(row)8990 print("=================== 12. 为了防止数据库查询发生 SQL 注入的攻击，我们可以使用 %s 占位符来转义查询的条件： ===================")91 sql = "SELECT * FROM sites WHERE name = %s"92 quer_name = ("Apple",)93 mycorsor.execute(sql, quer_name)94 query_all = mycorsor.fetchall()95 for row in query_all:96 print(row)9798 print("=================== 13.排序 ===================")99 sql = "SELECT * FROM sites ORDER BY name "100 mycorsor.execute(sql)101 query_all = mycorsor.fetchall()102 for row in query_all:103 print(row)104105 print("=================== 13.按 name 字段字母的降序排序 ===================")106 sql = "SELECT * FROM sites ORDER BY name DESC"107 mycorsor.execute(sql)108 query_all = mycorsor.fetchall()109 for row in query_all:110 print(row)111112 print("=================== 14.如果我们要设置查询的数据量，可以通过 LIMIT 语句来指定 ===================")113 # sql = "SELECT * FROM sites LIMIT 3"114 sql = "SELECT * FROM sites ORDER BY RAND() LIMIT 5"115 mycorsor.execute(sql)116 query_all = mycorsor.fetchall()117 for row in query_all:118 print(row)119120 # 随机查询121 print("=================== 14.随机查询1条 ===================")122 # sql = "SELECT * FROM sites WHERE id >= (SELECT FLOOR( MAX(id) * RAND()) FROM sites )ORDER BY id LIMIT 1;"123 sql = "SELECT * FROM sites WHERE id >= (SELECT floor(RAND() * (SELECT MAX(id) FROM sites))) ORDER BY id LIMIT 1;"124 mycorsor.execute(sql)125 query_all = mycorsor.fetchall()126 for row in query_all:127 print(row)128129 print("=================== 15.从第二条开始读取前 3 条记录： ===================")130 sql = "SELECT * FROM sites LIMIT 3 OFFSET 1"131 mycorsor.execute(sql)132 query_all = mycorsor.fetchall()133 for row in query_all:134 print(row)135136 print("=================== 16.删除记录 ===================")137 sql = "DELETE FROM sites WHERE name = 'apple'"138 mycorsor.execute(sql)139 mydb.commit()140 delete_count = mycorsor.rowcount141 print("{}条记录删除".format(delete_count))142143 print("=================== 17.为了防止数据库查询发生 SQL 注入的攻击，我们可以使用 %s 占位符来转义删除语句的条件： ===================")144 sql = "DELETE FROM sites WHERE name = %s"145 values = ("baidu",)146 mycorsor.execute(sql, values)147 mydb.commit()148 delete_count = mycorsor.rowcount149 print("{}条记录删除".format(delete_count))150151 print("=================== 18.更新表数据 ===================")152 sql = "UPDATE sites SET NAME = 'QQ' WHERE name = 'Tencent'"153 mycorsor.execute(sql)154 mydb.commit()155 update_count = mycorsor.rowcount156 print("{}条记录被更新".format(update_count))157158 print("=================== 18.更新表数据 ===================")159 sql = "UPDATE sites SET NAME = %s WHERE name = %s"160 values = ("Github", "github")161 mycorsor.execute(sql, values)162 mydb.commit()163 update_count = mycorsor.rowcount164 print("{}条记录被更新".format(update_count))165166 print("=================== 19.删除表 ===================")167 sql = "DROP TABLE IF EXISTS sites" ...

Full Screen

Full Screen

threadlib.py

Source:threadlib.py Github

copy

Full Screen

...22### END OF MODULE IMPORTS ###2324### START OF GLOBAL VARIABLES DECLARATION ###25# --------------------------------------------------------------- #26ALL_SERVERS_FROM_DB = ServersManipulator(Server().query_all())27ALL_SERVERS_CHANGES = ServerChangeLogManipulator(ServerChanges().query_all())28ALL_USERS_FROM_DB = UsersManipulator(User().query_all())29ALL_USER_CHANGES = UserChangeLogManipulator(UserChanges().query_all())30ALL_RPMS_FROM_DB = RPMsManipulator(RPM().query_all())31ALL_RPM_CHANGES = RPMChangeLogManipulator(RPMChanges().query_all())32# --------------------------------------------------------------- #33### END OF GLOBAL VARIABLES DECLARATION ###343536### START OF UTILITY FUNCTIONS DECLARATION37# --------------------------------------------------------------- #38def set_global_vars():39 while True:40 global ALL_SERVERS_FROM_DB, ALL_SERVERS_CHANGES, \41 ALL_USERS_FROM_DB, ALL_USER_CHANGES, \42 ALL_RPMS_FROM_DB, ALL_RPM_CHANGES43 ALL_SERVERS_FROM_DB = ServersManipulator(Server().query_all())44 ALL_SERVERS_CHANGES = ServerChangeLogManipulator(ServerChanges().query_all())45 ALL_USERS_FROM_DB = UsersManipulator(User().query_all())46 ALL_USER_CHANGES = UserChangeLogManipulator(UserChanges().query_all())47 ALL_RPMS_FROM_DB = RPMsManipulator(RPM().query_all())48 ALL_RPM_CHANGES = RPMChangeLogManipulator(RPMChanges().query_all())49 time.sleep(60)50# --------------------------------------------------------------- #51### END OF UTILITY FUNCTIONS DECLARATION525354def start_dbquery_thread():55 set_vars_thread = threading.Thread(target=set_global_vars)56 # set_vars_thread.name = time.time() ...

Full Screen

Full Screen

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 robotframework-ioslibrary automation tests on LambdaTest cloud grid

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

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful