How to use drop_unique_index method in autotest

Best Python code snippet using autotest_python

managers.py

Source:managers.py Github

copy

Full Screen

...35 def __init__(self, db_name):36 self.db_name = db_name37 def create_unique_index(self, key):38 pass39 def drop_unique_index(self, key):40 pass41class Mysql57IndexManager(BaseIndexManager):42 def create_unique_index(self, key):43 index_name = "{0}_contents_{1}".format(self.db_name, key)44 table_name = "worksheet_{}".format(self.db_name)45 # 新增一列46 add_column_sql = "ALTER TABLE {0} ADD COLUMN {1} varchar(128) GENERATED ALWAYS AS (contents->'$.{2}');".format(47 table_name, key, key48 )49 # 创建一个索引50 create_index_sql = "CREATE UNIQUE INDEX {0} ON {1}({2});".format(51 index_name, table_name, key52 )53 start = time.time()54 logger.info("start: {}".format(start))55 logger.info("add_column_sql is {}".format(add_column_sql))56 logger.info("create_index_sql is {}".format(create_index_sql))57 try:58 cursor = connection.cursor()59 logger.info("get cursor: {}".format(time.time() - start))60 cursor.execute(add_column_sql)61 logger.info("execute add_column_sql: {}".format(time.time() - start))62 cursor.execute(create_index_sql)63 logger.info("execute create_index_sql: {}".format(time.time() - start))64 except IntegrityError as e:65 logger.info("数据库索引创建失败,已回滚,db_name={}, key={}".format(self.db_name, key))66 self.drop_unique_index(key)67 raise e68 def drop_unique_index(self, key):69 table_name = "worksheet_{}".format(self.db_name)70 # 删除一列71 delete_column_sql = "ALTER TABLE {0} DROP COLUMN {1};".format(table_name, key)72 cursor = connection.cursor()73 cursor.execute(delete_column_sql)74class Mysql8IndexManager(BaseIndexManager):75 def create_unique_index(self, key):76 index_name = "{0}_contents_{1}".format(self.db_name, key)77 table_name = "worksheet_{}".format(self.db_name)78 sql = "CREATE UNIQUE INDEX {0} ON {1}( ( cast( contents->>'$.{2}' as char(128) ) ) );".format(79 index_name, table_name, key80 )81 # 真正的原生sql,82 cursor = connection.cursor()83 cursor.execute(sql)84 def drop_unique_index(self, key):85 index_name = "{0}_contents_{1}".format(self.db_name, key)86 table_name = "worksheet_{}".format(self.db_name)87 sql = "DROP INDEX {0} on {1};".format(index_name, table_name)88 cursor = connection.cursor()89 cursor.execute(sql)90class BaseTableManager:91 def __init__(self, table_name):92 self.table_name = table_name93 def drop_table(self):94 pass95class TableManager(BaseTableManager):96 def drop_table(self):97 sql = "DROP TABLE {};".format(self.table_name)98 cursor = connection.cursor()99 cursor.execute(sql)100class DataManager:101 def __init__(self, worksheet_id):102 self.worksheet_id = worksheet_id103 self.index_manager = Mysql57IndexManager104 if CURRENT_MYSQL_VERSION == 8:105 self.index_manager = Mysql8IndexManager106 @property107 def worksheet(self):108 return self.handler.get_instance()109 @property110 def handler(self):111 return WorkSheetHandler(worksheet_id=self.worksheet_id)112 @property113 def fields(self):114 return self.handler.get_fields()115 @property116 def db_name(self):117 return "{0}_{1}".format(self.worksheet.project_key.lower(), self.worksheet.key)118 @property119 def model(self):120 return DjangoBackend(self.db_name).get_model()121 def serializer_validate(self, data):122 serializer_class = SerializersGenerator(self.fields).build_serializers()(123 data=data124 )125 serializer_class.is_valid(raise_exception=True)126 validated_data = serializer_class.validated_data127 return validated_data128 def get_queryset(self):129 return self.model.objects.all()130 def get(self, pk):131 return self.model.objects.get(id=pk)132 def add(self, data, creator=None):133 validated_data = self.serializer_validate(data)134 with transaction.atomic():135 # 生成一个对象136 obj = self.model()137 obj.contents = validated_data138 obj.creator = creator139 obj.save()140 # 生成编号141 data = self.add_auto_number(validated_data)142 if data:143 obj.contents.update(data)144 obj.save()145 # 生成评分146 data = self.add_auto_grade(validated_data)147 if data:148 obj.contents.update(data)149 obj.save()150 # 存在公式控件,数值计算151 data = self.add_formula_result(validated_data)152 if data:153 obj.contents.update(data)154 obj.save()155 return obj156 def add_auto_number(self, validated_data):157 data = {}158 fields = self.get_auto_number_fields()159 for field in fields:160 data[field["key"]] = self.generate_number(161 field, validated_data=validated_data162 )163 return data164 def add_auto_grade(self, validated_data):165 data = {}166 fields = self.get_auto_grade_fields()167 for field in fields:168 data[field["key"]] = self.generate_grade(169 field, validated_data=validated_data170 )171 return data172 def get_auto_grade_fields(self):173 return [field for field in self.fields if field["type"] == "GRADE"]174 def generate_grade(self, field, validated_data):175 return GradeGenerator(field).generate_grade(validated_data=validated_data)176 def generate_number(self, field, validated_data):177 return NumberGeneratorDispatcher(178 field, validated_data=validated_data179 ).generate_number()180 def get_auto_number_fields(self):181 return [field for field in self.fields if field["type"] == "AUTO-NUMBER"]182 def add_formula_result(self, validated_data):183 data = {}184 fields = self.get_formula_fields()185 if not fields:186 return data187 for field in fields:188 result = self.generate_formula_result(field, validated_data)189 data.setdefault(field["key"], result)190 return data191 def get_formula_fields(self):192 return [field for field in self.fields if field["type"] == "FORMULA"]193 def generate_formula_result(self, field, validated_data):194 return FormulaGenerator(field).generate_formula_result(validated_data)195 def delete(self, pk):196 with transaction.atomic():197 # 生成一个对象198 self.model.objects.get(id=pk).delete()199 def update(self, pk, data, updated_by=None):200 validated_data = self.serializer_validate(data)201 with transaction.atomic():202 # 生成一个对象203 obj = self.model.objects.get(id=pk)204 obj.contents = self.update_content(obj.contents, validated_data)205 obj.updated_by = updated_by206 obj.save()207 def update_content(self, obj_contents, validated_data):208 # content数据更新209 contents = copy.deepcopy(obj_contents)210 # 兼容隐藏字段直接采用原数据211 for key, value in validated_data.items():212 if not value:213 continue214 contents[key] = value215 # 计算控件重新计算216 data = self.add_formula_result(contents)217 if data:218 contents.update(data)219 return contents220 def create_unique_index(self, key):221 self.index_manager(self.db_name).create_unique_index(key)222 def drop_unique_index(self, key):223 self.index_manager(self.db_name).drop_unique_index(key)224 def drop_table(self):225 table_name = "worksheet_{}".format(self.db_name)226 TableManager(table_name).drop_table()227class SerializersGenerator:228 def __init__(self, fields):229 self.fields = fields230 def build_serializers(self):231 attrs = {}232 for filed in self.fields:233 serializer = SerializerDispatcher(filed).get_serializer()234 attrs[filed["key"]] = serializer...

Full Screen

Full Screen

037_db_constraints.py

Source:037_db_constraints.py Github

copy

Full Screen

...42 # reasons43 old_index_name = basic_index_name(pivot_table, first_field)44 execute_safely(manager, 'DROP INDEX %s ON %s' %45 (old_index_name, pivot_table))46def drop_unique_index(manager, pivot_table, first_field):47 index_name = unique_index_name(pivot_table)48 manager.execute('DROP INDEX %s ON %s' % (index_name, pivot_table))49 old_index_name = basic_index_name(pivot_table, first_field)50 manager.execute('CREATE INDEX %s ON %s (%s)' %51 (old_index_name, pivot_table, first_field))52def foreign_key_name(table, field):53 return '_'.join([table, field, 'fk'])54def create_foreign_key_constraint(manager, table, field, destination_table):55 key_name = foreign_key_name(table, field)56 manager.execute('ALTER TABLE %s ADD CONSTRAINT %s FOREIGN KEY (%s) '57 'REFERENCES %s (id) ON DELETE NO ACTION' %58 (table, key_name, field, destination_table))59def drop_foreign_key_constraint(manager, table, field):60 key_name = foreign_key_name(table, field)61 manager.execute('ALTER TABLE %s DROP FOREIGN KEY %s' % (table, key_name))62def cleanup_m2m_pivot(manager, pivot_table, first_field, first_table,63 second_field, second_table, create_unique):64 delete_duplicates(manager, pivot_table, first_field, second_field)65 delete_invalid_foriegn_keys(manager, pivot_table, first_field, first_table)66 delete_invalid_foriegn_keys(manager, pivot_table, second_field,67 second_table)68 if create_unique:69 # first field is the more commonly used one, so we'll replace the70 # less-commonly-used index with the larger unique index71 create_unique_index(manager, pivot_table, second_field, first_field)72 create_foreign_key_constraint(manager, pivot_table, first_field,73 first_table)74 create_foreign_key_constraint(manager, pivot_table, second_field,75 second_table)76def reverse_cleanup_m2m_pivot(manager, pivot_table, first_field, second_field,77 drop_unique):78 drop_foreign_key_constraint(manager, pivot_table, second_field)79 drop_foreign_key_constraint(manager, pivot_table, first_field)80 if drop_unique:81 drop_unique_index(manager, pivot_table, second_field)82TABLES = (83 ('hosts_labels', 'host_id', 'hosts', 'label_id', 'labels', True),84 ('acl_groups_hosts', 'host_id', 'hosts', 'aclgroup_id', 'acl_groups',85 True),86 ('acl_groups_users', 'user_id', 'users', 'aclgroup_id', 'acl_groups',87 True),88 ('autotests_dependency_labels', 'test_id', 'autotests', 'label_id',89 'labels', False),90 ('jobs_dependency_labels', 'job_id', 'jobs', 'label_id', 'labels',91 False),92 ('ineligible_host_queues', 'job_id', 'jobs', 'host_id', 'hosts', True),93 )94def migrate_up(manager):95 for (table, first_field, first_table, second_field, second_table,...

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 autotest 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