Enable flake8-simplify plugin for ruff (#2328)

Fixed:

* SIM118 Use `key in dict` instead of `key in dict.keys()`
* SIM300 Yoda condition detected
* SIM105 Use `contextlib.suppress(...)` instead of `try`-`except`-`pass`
* SIM103 Return the condition directly
* SIM102 Use a single `if` statement instead of nested `if` statements
* SIM112 Use capitalized environment variable
* SIM114 Combine `if` branches using logical `or` operator
* SIM910 Use `dict.get(key)` instead of `dict.get(key, None)`
* SIM401 Use `self.get(key, default)` instead of an `if` block
* SIM212 `if`-expression with twisted arms
* SIM110 Reimplemented builtin
* SIM115 Use a context manager for opening files
* SIM117 Use a single `with` statement with multiple contexts instead of nested `with` statements

Ignored:

* SIM108 Use ternary operator instead of `if`-`else`-block

Add noqa for `SIM113` and `SIM116`

Statistics:

31      SIM118  [*] in-dict-keys
26      SIM105  [ ] suppressible-exception
13      SIM108  [*] if-else-block-instead-of-if-exp
10      SIM103  [ ] needless-bool
10      SIM300  [*] yoda-conditions
 9      SIM102  [ ] collapsible-if
 8      SIM112  [ ] uncapitalized-environment-variables
 7      SIM115  [ ] open-file-with-context-handler
 7      SIM117  [ ] multiple-with-statements
 5      SIM110  [*] reimplemented-builtin
 2      SIM113  [ ] enumerate-for-loop
 1      SIM114  [*] if-with-same-arms
 1      SIM116  [ ] if-else-block-instead-of-dict-lookup
 1      SIM212  [*] if-expr-with-twisted-arms
 1      SIM401  [*] if-else-block-instead-of-dict-get
 1      SIM910  [*] dict-get-with-none-default

dev/common/tdd.py

@@ -60,16 +60,10 @@ def verify_test_files_changed(changed_files):
     ]
 
     def is_app_path(fn):
-        for ap in app_paths:
-            if fn.startswith(ap):
-                return True
-        return False
+        return any(fn.startswith(ap) for ap in app_paths)
 
     def is_test_path(fn):
-        for tp in test_paths:
-            if fn.startswith(tp):
-                return True
-        return False
+        return any(fn.startswith(tp) for tp in test_paths)
 
     # exit early if no non-test changed in the api code
     app_changed = False

galaxy_ng/app/access_control/access_policy.py

@@ -1,3 +1,4 @@
+import contextlib
 import logging
 import os
 
@@ -224,14 +225,12 @@ def scope_queryset(self, view, qs):
         Scope the queryset based on the access policy `scope_queryset` method if present.
         """
         access_policy = self.get_access_policy(view)
-        if view.action == "list" and access_policy:
-            # if access_policy := self.get_access_policy(view):
-            if access_policy.queryset_scoping:
-                scope = access_policy.queryset_scoping["function"]
-                if scope == "scope_queryset" or not (func := getattr(self, scope, None)):
-                    return qs
-                kwargs = access_policy.queryset_scoping.get("parameters") or {}
-                qs = func(view, qs, **kwargs)
+        if view.action == "list" and access_policy and access_policy.queryset_scoping:
+            scope = access_policy.queryset_scoping["function"]
+            if scope == "scope_queryset" or not (func := getattr(self, scope, None)):
+                return qs
+            kwargs = access_policy.queryset_scoping.get("parameters") or {}
+            qs = func(view, qs, **kwargs)
         return qs
 
     # Define global conditions here
@@ -286,7 +285,7 @@ def v3_can_destroy_collections(self, request, view, action):
             return True
 
         # check collection object level permissions ...
-        if user.has_perm("ansible.delete_collection", collection):
+        if user.has_perm("ansible.delete_collection", collection):  # noqa: SIM103
             return True
 
         return False
@@ -303,7 +302,7 @@ def v3_can_view_users(self, request, view, action):
         if is_github_social_auth:
             return True
 
-        if request.user.has_perm('galaxy.view_user'):
+        if request.user.has_perm('galaxy.view_user'):  # noqa: SIM103
             return True
 
         return False
@@ -550,21 +549,16 @@ def is_not_protected_base_path(self, request, view, action):
                 base_path__in=PROTECTED_BASE_PATHS,
             ).exists():
                 return False
-        elif isinstance(obj, core_models.Distribution):
-            if obj.base_path in PROTECTED_BASE_PATHS:
-                return False
+        elif isinstance(obj, core_models.Distribution) and obj.base_path in PROTECTED_BASE_PATHS:
+            return False
 
         return True
 
     def require_requirements_yaml(self, request, view, action):
 
         if remote := request.data.get("remote"):
-            try:
+            with contextlib.suppress(ansible_models.CollectionRemote.DoesNotExist):
                 remote = ansible_models.CollectionRemote.objects.get(pk=extract_pk(remote))
-
-            except ansible_models.CollectionRemote.DoesNotExist:
-                pass
-
         if not remote:
             obj = view.get_object()
             remote = obj.remote.cast()
@@ -828,7 +822,7 @@ def is_namespace_owner(self, request, viewset, action):
 
         # use the helper to get the list of owners
         owners = get_v3_namespace_owners(v3_namespace)
-        if owners and user in owners:
+        if owners and user in owners:  # noqa: SIM103
             return True
 
         return False

galaxy_ng/app/access_control/statements/insights.py

@@ -149,7 +149,7 @@ def _entitelify(policy):
             # don't set conditions on deny statements. Otherwise, that will make it so
             # that users will only get denied if they have entitleements.
             if new_statement["effect"] == "allow":
-                condition = new_statement.get("condition", None)
+                condition = new_statement.get("condition")
 
                 if condition is None:
                     new_statement["condition"] = ["has_rh_entitlements"]

galaxy_ng/app/api/ui/v1/serializers/collection.py

@@ -27,7 +27,7 @@ def to_representation(self, contents):
     @staticmethod
     def _get_content_type_key(content_type: str) -> str:
         # FIXME(cutwater): Replace with galaxy-importer constants usage
-        if content_type == "role":  # ContentType.ROLE (from galaxy-importer)
+        if content_type == "role":  # ContentType.ROLE (from galaxy-importer)  # noqa: SIM116
             return "role"
         elif content_type == "module":  # ContentType.MODULE (from galaxy-importer)
             return "module"

galaxy_ng/app/api/ui/v1/serializers/execution_environment.py

@@ -185,6 +185,4 @@ def get_write_only_fields(self, obj):
         return utils.get_write_only_fields(self, obj)
 
     def get_is_indexable(self, obj) -> bool:
-        if obj.get_registry_backend():
-            return True
-        return False
+        return bool(obj.get_registry_backend())

galaxy_ng/app/api/ui/v2/permissions.py

@@ -16,7 +16,7 @@ def has_permission(self, request, view):
         if request.user and request.user.is_superuser:
             return True
         # Allow read-only access (GET, HEAD, OPTIONS) for other users
-        if request.method in SAFE_METHODS:
+        if request.method in SAFE_METHODS:  # noqa: SIM103
             return True
         # Otherwise, deny access
         return False
@@ -104,7 +104,7 @@ def has_object_permission(self, request, view, obj):
             return True
 
         # user can set themself only to False
-        if request.user == obj and request.data.get('is_superuser') is False:
+        if request.user == obj and request.data.get('is_superuser') is False:  # noqa: SIM103
             return True
 
         return False

galaxy_ng/app/api/ui/v2/serializers.py

@@ -117,7 +117,7 @@ def validate_groups(self, groups):
         if groups is not None:
             for group_dict in groups:
                 group_filter = {}
-                for field in group_dict.keys():
+                for field in group_dict:
                     if field in ('id', 'name'):
                         group_filter[field] = group_dict[field]
 
@@ -136,7 +136,7 @@ def validate_teams(self, teams):
         if teams is not None:
             for team_dict in teams:
                 team_filter = {}
-                for field in team_dict.keys():
+                for field in team_dict:
                     if field in ('id', 'name'):
                         team_filter[field] = team_dict[field]
                 try:
@@ -154,7 +154,7 @@ def validate_organizations(self, organizations):
         if organizations is not None:
             for org_dict in organizations:
                 org_filter = {}
-                for field in org_dict.keys():
+                for field in org_dict:
                     if field in ('id', 'name'):
                         org_filter[field] = org_dict[field]
                 try:

galaxy_ng/app/api/v1/serializers.py

@@ -351,9 +351,8 @@ def get_summary_fields(self, obj):
 
         # prefer the provider avatar url
         avatar_url = f'https://github.com/{obj.namespace.name}.png'
-        if obj.namespace and obj.namespace.namespace:
-            if obj.namespace.namespace.avatar_url:
-                avatar_url = obj.namespace.namespace.avatar_url
+        if obj.namespace and obj.namespace.namespace and obj.namespace.namespace.avatar_url:
+            avatar_url = obj.namespace.namespace.avatar_url
 
         return {
             'dependencies': dependencies,

galaxy_ng/app/api/v1/tasks.py

@@ -1,3 +1,4 @@
+import contextlib
 import copy
 import datetime
 import logging
@@ -314,10 +315,8 @@ def legacy_role_import(
     v1_task_id = None
     task_id = None
     import_model = None
-    try:
+    with contextlib.suppress(Exception):
         task = Task.current()
-    except Exception:
-        pass
     if task:
         v1_task_id = uuid_to_int(str(task.pulp_id))
         task_id = task.pulp_id

galaxy_ng/app/api/v3/viewsets/collection.py

@@ -182,7 +182,7 @@ def get(self, request, *args, **kwargs):
         if settings.get("ANSIBLE_COLLECT_DOWNLOAD_COUNT", False):
             pulp_ansible_views.CollectionArtifactDownloadView.count_download(filename)
 
-        if settings.GALAXY_DEPLOYMENT_MODE == DeploymentMode.INSIGHTS.value:
+        if settings.GALAXY_DEPLOYMENT_MODE == DeploymentMode.INSIGHTS.value:  # noqa: SIM300
             url = 'http://{host}:{port}/{prefix}/{distro_base_path}/{filename}'.format(
                 host=settings.X_PULP_CONTENT_HOST,
                 port=settings.X_PULP_CONTENT_PORT,
@@ -211,7 +211,7 @@ def get(self, request, *args, **kwargs):
             raise APIException(
                 _('Unexpected response from content app. Code: %s.') % response.status_code
             )
-        elif settings.GALAXY_DEPLOYMENT_MODE == DeploymentMode.STANDALONE.value:
+        elif settings.GALAXY_DEPLOYMENT_MODE == DeploymentMode.STANDALONE.value:  # noqa: SIM300
             url = '{host}/{prefix}/{distro_base_path}/{filename}'.format(
                 host=settings.CONTENT_ORIGIN.strip("/"),
                 prefix=prefix,
@@ -321,25 +321,28 @@ def move_content(self, request, *args, **kwargs):
             move_task = call_sign_and_move_task(signing_service, **move_task_params)
         else:
             require_signatures = settings.get("GALAXY_REQUIRE_SIGNATURE_FOR_APPROVAL", False)
-            if dest_repo.name == golden_repo and require_signatures:
-                if collection_version.signatures.count() == 0:
-                    return Response(
-                        {
-                            "detail": _(
-                                "Collection {namespace}.{name} could not be approved "
-                                "because system requires at least a signature for approval."
-                            ).format(
-                                namespace=collection_version.namespace,
-                                name=collection_version.name,
-                            )
-                        },
-                        status=status.HTTP_400_BAD_REQUEST,
-                    )
+            if (
+                dest_repo.name == golden_repo
+                and require_signatures
+                and collection_version.signatures.count() == 0
+            ):
+                return Response(
+                    {
+                        "detail": _(
+                            "Collection {namespace}.{name} could not be approved "
+                            "because system requires at least a signature for approval."
+                        ).format(
+                            namespace=collection_version.namespace,
+                            name=collection_version.name,
+                        )
+                    },
+                    status=status.HTTP_400_BAD_REQUEST,
+                )
             move_task = call_move_content_task(**move_task_params)
 
         response_data['copy_task_id'] = response_data['remove_task_id'] = move_task.pk
 
-        if settings.GALAXY_DEPLOYMENT_MODE == DeploymentMode.INSIGHTS.value:
+        if settings.GALAXY_DEPLOYMENT_MODE == DeploymentMode.INSIGHTS.value:  # noqa: SIM300
             golden_repo = AnsibleDistribution.objects.get(
                 base_path=settings.GALAXY_API_DEFAULT_DISTRIBUTION_BASE_PATH
             ).repository

galaxy_ng/app/auth/ldap.py

@@ -53,11 +53,9 @@ def authenticate(self, *args, **kwargs):
         When a prefixed user authenticates, remove the prefix from their
         username kwarg and treat them as the non-prefixed user.
         """
-        if username := kwargs.get("username"):
-            if username.startswith(self.prefix):
-                kwargs["username"] = username.removeprefix(self.prefix)
-                return super().authenticate(*args, **kwargs)
-
+        username = kwargs.get("username")
+        if username and username.startswith(self.prefix):
+            kwargs["username"] = username.removeprefix(self.prefix)
         return super().authenticate(*args, **kwargs)
 
     def get_or_build_user(self, username, ldap_user):

galaxy_ng/app/dynaconf_hooks.py

@@ -426,14 +426,13 @@ def configure_authentication_classes(settings: Dynaconf, data: Dict[str, Any]) -
 
     # galaxy sessionauth -must- always come first ...
     galaxy_session = "galaxy_ng.app.auth.session.SessionAuthentication"
-    if galaxy_auth_classes:
-        # Check if galaxy_session is already the first element
-        if galaxy_auth_classes[0] != galaxy_session:
-            # Remove galaxy_session if it exists in the list
-            if galaxy_session in galaxy_auth_classes:
-                galaxy_auth_classes.remove(galaxy_session)
-            # Insert galaxy_session at the beginning of the list
-            galaxy_auth_classes.insert(0, galaxy_session)
+    # Check if galaxy_session is already the first element
+    if galaxy_auth_classes and galaxy_auth_classes[0] != galaxy_session:
+        # Remove galaxy_session if it exists in the list
+        if galaxy_session in galaxy_auth_classes:
+            galaxy_auth_classes.remove(galaxy_session)
+        # Insert galaxy_session at the beginning of the list
+        galaxy_auth_classes.insert(0, galaxy_session)
 
     if galaxy_auth_classes:
         data["ANSIBLE_AUTHENTICATION_CLASSES"] = list(galaxy_auth_classes)

galaxy_ng/app/management/commands/sync-galaxy-collections.py

@@ -67,7 +67,7 @@ def handle(self, *args, **options):
             collection_name=options['name'],
             limit=options['limit'],
         ):
-            counter += 1
+            counter += 1  # noqa: SIM113
             logger.info(
                 f"{counter}. {collection_info['namespace']['name']}.{collection_info['name']}"
                 + f" versions:{len(collection_versions)}"

galaxy_ng/app/management/commands/sync-galaxy-namespaces.py

@@ -54,7 +54,7 @@ def handle(self, *args, **options):
                 limit=options['limit'],
             ):
 
-                count += 1
+                count += 1  # noqa: SIM113
                 namespace_name = namespace_info['name']
 
                 self.echo(

galaxy_ng/app/metrics_collection/automation_analytics/data.py

@@ -116,10 +116,9 @@ def _simple_csv(full_path, file_name, query, max_data_size=209715200):
     file_path = _get_file_path(full_path, file_name)
     tfile = _get_csv_splitter(file_path, max_data_size)
 
-    with connection.cursor() as cursor:
-        with cursor.copy(query) as copy:
-            while data := copy.read():
-                tfile.write(str(data, 'utf8'))
+    with connection.cursor() as cursor, cursor.copy(query) as copy:
+        while data := copy.read():
+            tfile.write(str(data, 'utf8'))
 
     return tfile.file_list()
 

galaxy_ng/app/metrics_collection/lightspeed/data.py

@@ -199,10 +199,9 @@ def _simple_csv(full_path, file_name, query, max_data_size=209715200):
     file_path = _get_file_path(full_path, file_name)
     tfile = _get_csv_splitter(file_path, max_data_size)
 
-    with connection.cursor() as cursor:
-        with cursor.copy(query) as copy:
-            while data := copy.read():
-                tfile.write(str(data, 'utf8'))
+    with connection.cursor() as cursor, cursor.copy(query) as copy:
+        while data := copy.read():
+            tfile.write(str(data, 'utf8'))
 
     return tfile.file_list()
 

galaxy_ng/app/metrics_collection/lightspeed/package.py

@@ -18,16 +18,16 @@ def get_ingress_url(self):
         return False
 
     def _get_rh_user(self):
-        return os.environ["aws_access_key_id"]
+        return os.environ["aws_access_key_id"]  # noqa: SIM112
 
     def _get_rh_password(self):
-        return os.environ["aws_secret_access_key"]
+        return os.environ["aws_secret_access_key"]  # noqa: SIM112
 
     def _get_rh_region(self):
-        return os.environ["aws_region"]
+        return os.environ["aws_region"]  # noqa: SIM112
 
     def _get_rh_bucket(self):
-        return os.environ["aws_bucket"]
+        return os.environ["aws_bucket"]  # noqa: SIM112
 
     def get_s3_configured(self):
         return True

galaxy_ng/app/models/namespace.py

@@ -65,7 +65,7 @@ class Namespace(
     def avatar_url(self):
         # TODO: remove this once we can fix the content app on CRC
         # the content app in crc doesn't work
-        if settings.GALAXY_DEPLOYMENT_MODE == DeploymentMode.STANDALONE.value:
+        if settings.GALAXY_DEPLOYMENT_MODE == DeploymentMode.STANDALONE.value:  # noqa: SIM300
             data = self.last_created_pulp_metadata
             if data and data.avatar_sha256:
                 return settings.ANSIBLE_API_HOSTNAME + get_url(data) + "avatar/"

galaxy_ng/app/signals/handlers.py

@@ -112,10 +112,7 @@ def _update_metadata():
         ns.set_links([{"name": x, "url": instance.links[x]} for x in instance.links])
         ns.save()
 
-    if created or ns_metadata is None:
-        _update_metadata()
-
-    elif ns.metadata_sha256 != instance.metadata_sha256:
+    if created or ns_metadata is None or ns.metadata_sha256 != instance.metadata_sha256:
         _update_metadata()
 
 

galaxy_ng/app/utils/namespaces.py

@@ -60,7 +60,7 @@ def validate_namespace_name(name):
         return False
     if len(name) < 2:
         return False
-    if name.startswith('_'):
+    if name.startswith('_'):  # noqa: SIM103
         return False
     return True