Implement SonarCloud style suggestions

Author: AlanCoding

ansible_base/rbac/management/__init__.py

@@ -31,16 +31,16 @@ def create_dab_permissions(app_config, verbosity=2, interactive=True, using=DEFA
         return
 
     try:
-        DABContentType = apps.get_model("dab_rbac", "DABContentType")
+        dab_ct_cls = apps.get_model("dab_rbac", "DABContentType")
     except LookupError:
         logger.warning('Skipping DAB RBAC type and permission creation since models are not available')
         return
 
-    if not router.allow_migrate_model(using, DABContentType):
+    if not router.allow_migrate_model(using, dab_ct_cls):
         # Uncommon case, code logic is using a replica database or something, unlikely to be relevant
         return
 
-    sync_DAB_permissions(verbosity=verbosity, using=using, apps=global_apps)
+    sync_dab_permissions(verbosity=verbosity, using=using, apps=global_apps)
 
 
 def is_safe_identifier(name: str) -> bool:
@@ -67,13 +67,13 @@ def reset_ct_sequence(ct_cls):
         )
 
 
-def sync_DAB_permissions(verbosity=2, using=DEFAULT_DB_ALIAS, apps=global_apps):
+def sync_dab_permissions(verbosity=2, using=DEFAULT_DB_ALIAS, apps=global_apps):
     """Idepotent method to set database types and permissions for DAB RBAC
 
     This should make the database content reflect the model Meta data and
     registrations in the permission_registry for that app.
     """
-    DABContentType = apps.get_model("dab_rbac", "DABContentType")
+    dab_ct_cls = apps.get_model("dab_rbac", "DABContentType")
     Permission = apps.get_model("dab_rbac", "DABPermission")
 
     new_cts = create_DAB_contenttypes(verbosity=verbosity, using=using, apps=apps)
@@ -86,7 +86,7 @@ def sync_DAB_permissions(verbosity=2, using=DEFAULT_DB_ALIAS, apps=global_apps):
         # Force looking up the content types in the current database
         # before creating foreign keys to them.
         service = get_resource_prefix(klass)
-        ctype = DABContentType.objects.db_manager(using).get_for_model(klass, service=service, for_concrete_model=False)
+        ctype = dab_ct_cls.objects.db_manager(using).get_for_model(klass, service=service, for_concrete_model=False)
 
         ctypes.add(ctype)
 
@@ -132,4 +132,4 @@ def sync_DAB_permissions(verbosity=2, using=DEFAULT_DB_ALIAS, apps=global_apps):
     # Reset the sequence to avoid PK collision later
     if connection.vendor == 'postgresql':
         if new_cts:
-            reset_ct_sequence(DABContentType)
+            reset_ct_sequence(dab_ct_cls)

ansible_base/rbac/management/create_types.py

@@ -31,31 +31,31 @@ def create_DAB_contenttypes(
 
     Returns a list of the _new_ content types created
     """
-    DABContentType = apps.get_model("dab_rbac", "DABContentType")
-    ContentType = apps.get_model("contenttypes", "ContentType")
+    dab_ct_cls = apps.get_model("dab_rbac", "DABContentType")
+    ct_cls = apps.get_model("contenttypes", "ContentType")
 
-    content_types = get_local_DAB_contenttypes(using, DABContentType)
+    content_types = get_local_DAB_contenttypes(using, dab_ct_cls)
 
     ct_data = []
     for model in permission_registry.all_registered_models:
         service = get_resource_prefix(model)
         if (service, model._meta.model_name) not in content_types:
             # The content type is not seen in existing entries, add to list for creation
-            ct_item_data = dict(
-                service=service,
-                app_label=model._meta.app_label,
-                model=model._meta.model_name,
-                api_slug=f'{service}.{model._meta.model_name}',
-                pk_field_type=model._meta.pk.db_type(connection),
-            )
+            ct_item_data = {
+                'service': service,
+                'app_label': model._meta.app_label,
+                'model': model._meta.model_name,
+                'api_slug': f'{service}.{model._meta.model_name}',
+                'pk_field_type': model._meta.pk.db_type(connection),
+            }
             # To make usage earier in a transitional period, we will set the content type
             # of any new entries created here to the id of its corresponding ContentType
             # from the actual contenttypes app, allowing many filters to work
-            real_ct = ContentType.objects.get_for_model(model)
-            if not DABContentType.objects.filter(id=real_ct.id).exists():
+            real_ct = ct_cls.objects.get_for_model(model)
+            if not dab_ct_cls.objects.filter(id=real_ct.id).exists():
                 ct_item_data['id'] = real_ct.id
             else:
-                current_max_id = DABContentType.objects.order_by('-id').values_list('id', flat=True).first() or 0
+                current_max_id = dab_ct_cls.objects.order_by('-id').values_list('id', flat=True).first() or 0
                 ct_item_data['id'] = current_max_id + 1
             ct_data.append(ct_item_data)
     if not ct_data:
@@ -66,19 +66,19 @@ def create_DAB_contenttypes(
     # from the actual contenttypes app, allowing many filters to work
     cts = []
     for ct_item_data in ct_data:
-        cts.append(DABContentType.objects.create(**ct_item_data))
+        cts.append(dab_ct_cls.objects.create(**ct_item_data))
 
     if verbosity >= 2:
         for ct in cts:
             logger.debug("Adding DAB content type " f"'{ct.service}:{ct.app_label} | {ct.model}'")
 
     updated_ct = 0
-    for ct in DABContentType.objects.all():
+    for ct in dab_ct_cls.objects.all():
         if not permission_registry.is_registered(ct.model_class()):
             logger.warning(f'{ct.model} is a stale content type in DAB RBAC')
             continue
         if parent_model := permission_registry.get_parent_model(ct.model_class()):
-            parent_content_type = DABContentType.objects.get_for_model(parent_model)
+            parent_content_type = dab_ct_cls.objects.get_for_model(parent_model)
             if ct.parent_content_type != parent_content_type:
                 ct.parent_content_type = parent_content_type
                 ct.save(update_fields=['parent_content_type'])

ansible_base/rbac/models/content_type.py

@@ -62,18 +62,21 @@ def get_for_model(
             return ct
 
         if service is None:
-            service = get_resource_prefix(model)
+            query_service = get_resource_prefix(model)
+        else:
+            query_service = service
+
         opts = self._get_opts(model, for_concrete_model)
         try:
-            return self._get_from_cache(opts, service)
+            return self._get_from_cache(opts, query_service)
         except KeyError:
             pass
 
         try:
-            ct = self.get(service=service, app_label=opts.app_label, model=opts.model_name)
+            ct = self.get(service=query_service, app_label=opts.app_label, model=opts.model_name)
         except self.model.DoesNotExist:
             raise RuntimeError(
-                f'Could not find content type for {(service, opts.app_label, opts.model_name)}, '
+                f'Could not find content type for {(query_service, opts.app_label, opts.model_name)}, '
                 'and creating new objects via get_for_model is not allowed for DAB RBAC'
             )
         self._add_to_cache(self.db, ct)
@@ -147,9 +150,8 @@ def get_by_natural_key(self, *args: str) -> django_models.Model:
             kwargs = {'service__in': [get_local_resource_prefix(), 'shared'], 'app_label': app_label, 'model': model}
             # This ask here is actually ambiguous, so we try this extra lookup
             shared_key = ('shared', app_label, model)
-            if self.db in self._cache:
-                if shared_key in self._cache[self.db]:
-                    return self._cache[self.db][shared_key]
+            if shared_key in self._cache.get(self.db, ()):
+                return self._cache[self.db][shared_key]
         else:
             service, app_label, model = args
             kwargs = {'service': service, 'app_label': app_label, 'model': model}