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

How to use get_public_schema_name method in Kiwi

Best Python code snippet using Kiwi_python

tasks.py

Source:tasks.py

...10    from django.db import connection11    from django_tenants.utils import get_public_schema_name, get_tenant_model12    from apps.classif_ai.services import InferenceService13    # Setting the correct schema14    if connection.schema_name != get_public_schema_name():15        connection.set_schema_to_public()16    if schema != get_public_schema_name():17        tenant = get_tenant_model().objects.get(schema_name=schema)18        connection.set_tenant(tenant, include_public=True)19    try:20        logger.info(f"Schema_name: {schema}, File Set Id: {file_set_id}, Model Id: {ml_model_id}")21        InferenceService(ml_model_id, file_set_id).perform()22    except ValidationError as e:23        # ToDo: justify this pass.24        if "file" in e.message_dict.keys():25            pass26        elif "FileSetInferenceQueue" in e.message_dict.keys():27            self.retry(exc=e, countdown=2**self.request.retries, max_retries=20)28        else:29            raise e30    return31@shared_task(bind=True, ignore_result=True)32def perform_retraining(self, training_session_id, schema="public"):33    from django.db import connection34    from django.db.models import Q35    from django_tenants.utils import get_public_schema_name, get_tenant_model36    from apps.classif_ai.models import TrainingSession, TrainingSessionFileSet, FileRegion37    from apps.classif_ai.helpers import calculate_iou38    from sixsense import settings39    sys.path.append(settings.DS_MODEL_INVOCATION_PATH)40    from retrain import main41    # Setting the correct schema42    if connection.schema_name != get_public_schema_name():43        connection.set_schema_to_public()44    if schema != get_public_schema_name():45        tenant = get_tenant_model().objects.get(schema_name=schema)46        connection.set_tenant(tenant, include_public=True)47    training_session = TrainingSession.objects.get(id=training_session_id)48    if training_session.started_at:49        return50    ml_model = training_session.new_ml_model51    training_session.started_at = datetime.utcnow()52    training_session.save()53    try:54        main(training_session_id, schema)55        training_session.refresh_from_db()56        training_session.finished_at = datetime.utcnow()57        training_session.save()58        ml_model.refresh_from_db()59    except Exception as e:60        ml_model.refresh_from_db()61        if ml_model.status != "user_terminated":62            ml_model.status = "training_failed"63            ml_model.save()64        training_session.refresh_from_db()65        training_session.finished_at = datetime.utcnow()66        training_session.save()67        raise e68    # ToDo: Copy the existing feedback as well for all the file sets69    batch_size = 100070    qs = TrainingSessionFileSet.objects.filter(71        Q(dataset_train_type="TEST") | Q(dataset_train_type="test"), training_session_id=training_session_id72    )73    total = qs.count()74    ml_model.status = "ready_for_deployment"75    ml_model.save()76    for start in range(0, total, batch_size):77        end = min(start + batch_size, total)78        for tfs in qs[start:end]:79            for file, regions in tfs.defects.items():80                ai_regions = FileRegion.objects.filter(81                    ml_model_id=training_session.new_ml_model_id, file_id=file, is_user_feedback=False82                )83                for region in regions:84                    matching_region = None85                    for ai_region in ai_regions:86                        if ml_model.type == "CLASSIFICATION":87                            if list(ai_region.defects)[0] == list(region["defects"])[0]:88                                matching_region = [ai_region, None]89                        else:90                            iou = calculate_iou(91                                [92                                    region["region"]["coordinates"]["x"],93                                    region["region"]["coordinates"]["y"],94                                    region["region"]["coordinates"]["x"] + region["region"]["coordinates"]["w"],95                                    region["region"]["coordinates"]["y"] + region["region"]["coordinates"]["h"],96                                ],97                                [98                                    ai_region.region["coordinates"]["x"],99                                    ai_region.region["coordinates"]["y"],100                                    ai_region.region["coordinates"]["x"] + ai_region.region["coordinates"]["w"],101                                    ai_region.region["coordinates"]["y"] + ai_region.region["coordinates"]["h"],102                                ],103                            )104                            if iou > 0.4:105                                if matching_region:106                                    if matching_region[1] < iou:107                                        matching_region = [ai_region, iou]108                                else:109                                    matching_region = [ai_region, iou]110                    file_region = FileRegion(111                        file_id=file,112                        ml_model_id=training_session.new_ml_model_id,113                        defects=region["defects"],114                        region=region["region"],115                        is_user_feedback=True,116                    )117                    if matching_region and ml_model.type == "CLASSIFICATION":118                        matching_region[0].classification_correctness = True119                        matching_region[0].save()120                    elif matching_region:121                        file_region.ai_region_id = matching_region[0].id122                        file_region.save()123                    else:124                        file_region.save()125                for ai_region in ai_regions.all():126                    if ml_model.type == "CLASSIFICATION" and ai_region.classification_correctness == None:127                        ai_region.classification_correctness = False128                        ai_region.save()129                    elif ml_model.type != "CLASSIFICATION" and ai_region.detection_correctness == None:130                        ai_region.is_removed = True131                        ai_region.classification_correctness = False132                        ai_region.detection_correctness = False133                        ai_region.save()134@shared_task(bind=True, ignore_result=True)135def stitch_image_worker(self, upload_session_id, schema="public"):136    from django.db import connection137    from django_tenants.utils import get_public_schema_name, get_tenant_model138    from apps.classif_ai.services import StitchImageService139    # Setting the correct schema140    if connection.schema_name != get_public_schema_name():141        connection.set_schema_to_public()142    if schema != get_public_schema_name():143        tenant = get_tenant_model().objects.get(schema_name=schema)144        connection.set_tenant(tenant, include_public=True)145    service = StitchImageService()146    service.stitch(upload_session_id)147def set_schema(schema):148    from django.db import connection149    from django_tenants.utils import get_public_schema_name, get_tenant_model150    if connection.schema_name != get_public_schema_name():151        connection.set_schema_to_public()152    if schema != get_public_schema_name():153        tenant = get_tenant_model().objects.get(schema_name=schema)154        connection.set_tenant(tenant, include_public=True)155# HOW to enforce first parameter needs to be description and second one needs to be input156@shared_task(bind=True)157def copy_images_to_folder(self, input_data, schema="public"):158    set_schema(schema)159    from apps.classif_ai.models import FileSet160    FileSet.objects.copy(input_data)161@before_task_publish.connect162def task_sent_handler(sender=None, headers=None, body=None, **kwargs):163    info = headers if "task" in headers else body164    if not info["ignore_result"]:165        from django.db import transaction166        from django_celery_results.models import TaskResult167        from django_tenants.utils import get_public_schema_name168        import ast169        schema = ast.literal_eval(info.get("kwargsrepr")).get("schema") or get_public_schema_name()170        set_schema(schema)171        with transaction.atomic():172            TaskResult.objects.get_or_create(173                content_type="application/json",174                content_encoding="utf-8",175                task_id=info["id"],176                status=states.PENDING,177                result=None,178                task_name=info["task"],179                task_args=info["argsrepr"],180                task_kwargs=info["kwargsrepr"],...

admin.py

Source:admin.py

...5class SharedAdmin(admin.ModelAdmin):6    @staticmethod7    def has_tenant(request):8        # print('current schema name: ', request.tenant.schema_name)9        # print('public schema name: ', get_public_schema_name())10        # public schema don't have any tenant11        if hasattr(request, "tenant"):12            return True13        else:14            return False15    def has_view_permission(self, request, view=None):16        return True17        # if self.has_tenant(request):18        #     # print('current schema name: ', request.tenant.schema_name)19        #     # print('public schema name: ', get_public_schema_name())20        #     if request.tenant.schema_name != get_public_schema_name():21        #         return False22    def has_add_permission(self, request, view=None):23        if self.has_tenant(request):24            if request.tenant.schema_name != get_public_schema_name():25                return False26        return True27    def has_change_permission(self, request, view=None):28        if self.has_tenant(request):29            if request.tenant.schema_name != get_public_schema_name():30                return False31        return True32    def has_delete_permission(self, request, view=None):33        if self.has_tenant(request):34            if request.tenant.schema_name != get_public_schema_name():35                return False36        return True37    # def has_view_or_change_permission(self, request, view=None):38    #     if self.has_tenant(request):39    #         if request.tenant.schema_name != get_public_schema_name():40    #             return False41    #     return True42depot_app_config = apps.get_app_config('shared_app')43for model in depot_app_config.get_models():44    admin.site.register(model, SharedAdmin)...

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.