Best Python code snippet using autotest_python
managers.py
Source:managers.py  
...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...037_db_constraints.py
Source:037_db_constraints.py  
...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,...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
