fix: apply ruff auto-fixes, enforce import ordering, fix logging (#1197)

Safe auto-fixes applied by `ruff check --fix`:
- I001: sort imports with custom section ordering
- F401: remove unused imports
- UP032: convert .format() to f-strings (non-logging calls only)
- W605: fix invalid escape sequences
- RET502/505/506: remove superfluous else after return/raise
- SIM114/117: combine if branches, merge with statements
- RUF010: fix f-string type conversion
- G010: logging.warn → logging.warning

Manual fixes:
- G004: convert f-strings in 27 logging calls to %-style
  for lazy evaluation

New rules added:
- G (flake8-logging-format): enforce %-style in logging calls

Import ordering: stdlib → third-party → pulpcore →
content plugins (pulp_rpm, pulp_npm, etc.) → pulp_service

Assisted-by: Claude Code <noreply@anthropic.com>

Author: decko

pulp_service/pulp_service/app/admin.py

@@ -1,24 +1,24 @@
-from django.contrib.auth.admin import UserAdmin, GroupAdmin
+import re
+
+from django import forms
+from django.contrib import admin
+from django.contrib.auth.admin import GroupAdmin, UserAdmin
 from django.contrib.auth.forms import AuthenticationForm, UserChangeForm, UserCreationForm
 from django.contrib.auth.models import User
-from django.contrib import admin
-from django.db.models import Q
 from django.core.exceptions import ValidationError
+from django.core.validators import RegexValidator
+from django.db.models import Q
 from django.urls import reverse
 from django.utils.html import format_html
-
-from django import forms
-from django.core.validators import RegexValidator
-
 from hijack.contrib.admin import HijackUserAdminMixin
 
-from .models import DomainOrg
-import re
-
-from pulpcore.plugin.models import Domain, Group
 from pulpcore.app.models import Task
+from pulpcore.plugin.models import Domain, Group
+
 from pulp_service.app.constants import CONTENT_SOURCES_LABEL_NAME
 
+from .models import DomainOrg
+
 USERNAME_PATTERN = r"^[\w.@+=/\-|]+$"
 USERNAME_ERROR_MSG = "Username can only contain letters, numbers, and these special characters: @, ., +, -, =, /, _, |"
 USERNAME_HELP_TEXT = (

pulp_service/pulp_service/app/authentication.py

@@ -1,14 +1,8 @@
-import jq
-import json
 import logging
 
-from base64 import b64decode
-from binascii import Error as Base64DecodeError
 from django.contrib.auth import get_user_model
-from gettext import gettext as _
-from pulpcore.app.authentication import JSONHeaderRemoteAuthentication
-from rest_framework.exceptions import AuthenticationFailed
 
+from pulpcore.app.authentication import JSONHeaderRemoteAuthentication
 
 _logger = logging.getLogger(__name__)
 
@@ -24,7 +18,7 @@ def authenticate_header(self, request):
 class RHTermsBasedRegistryAuthentication(JSONHeaderRemoteAuthentication):
     header = "HTTP_X_RH_IDENTITY"
     # Combines org_id with username - falls back to "|username" if org_id is missing
-    jq_filter = '.identity | if .user.username then "\(.org_id // "")|\(.user.username)" else null end'
+    jq_filter = r'.identity | if .user.username then "\(.org_id // "")|\(.user.username)" else null end'
 
     def authenticate_header(self, request):
         return "Bearer"
@@ -47,7 +41,7 @@ class TurnpikeTermsBasedRegistryAuthentication(JSONHeaderRemoteAuthentication):
     header = "HTTP_X_RH_IDENTITY"
     jq_filter = (
         'if .identity.auth_type == "registry-auth" and .identity.registry.username '
-        'then "\(.identity.registry.org_id // "")|\(.identity.registry.username)" '
+        r'then "\(.identity.registry.org_id // "")|\(.identity.registry.username)" '
         "else null end"
     )
 
@@ -73,5 +67,5 @@ def get_user(self, user_id):
         try:
             return User.objects.get(pk=user_id)
         except User.DoesNotExist:
-            _logger.warning(f"User with id {user_id} not found in get_user()")
+            _logger.warning("User with id %s not found in get_user()", user_id)
             return None

pulp_service/pulp_service/app/authorization.py

@@ -1,17 +1,16 @@
-from contextvars import ContextVar
+import json
+import logging
 from base64 import b64decode
 from binascii import Error as Base64DecodeError
-from django.db.models import Q
-
+from contextvars import ContextVar
 
-import json
 import jq
-from pulpcore.plugin.models import Domain
+from django.db.models import Q
+from rest_framework.permissions import SAFE_METHODS, BasePermission
+
 from pulpcore.plugin.util import extract_pk, get_domain_pk
-from pulp_service.app.models import DomainOrg
-from rest_framework.permissions import BasePermission, SAFE_METHODS
-import logging
 
+from pulp_service.app.models import DomainOrg
 
 _logger = logging.getLogger(__name__)
 org_id_var = ContextVar("org_id")
@@ -73,13 +72,9 @@ def has_permission(self, request, view):
             user_id_var.set(request.user.pk)
             return True
         # Anyone can list domains
-        elif action == "domain_list":
+        if action == "domain_list":
             return True
-        elif action == "domain_update":
-            # The PK is part of the URL
-            domain_pk = extract_pk(request.META["PATH_INFO"])
-            return self._has_domain_access(domain_pk, org_id, user)
-        elif action == "domain_delete":
+        if action == "domain_update" or action == "domain_delete":
             # The PK is part of the URL
             domain_pk = extract_pk(request.META["PATH_INFO"])
             return self._has_domain_access(domain_pk, org_id, user)
@@ -116,15 +111,15 @@ def get_decoded_identity_header(self, request):
                 header_decoded_content = b64decode(header_content)
                 return header_decoded_content
         except Base64DecodeError:
-            return
+            return None
 
     def get_org_id(self, decoded_header_content):
         if decoded_header_content:
             try:
                 header_value = json.loads(decoded_header_content)
                 return org_id_json_path.input_value(header_value).first()
             except json.JSONDecodeError:
-                return
+                return None
 
 
 class AllowUnauthPull(BasePermission):

pulp_service/pulp_service/app/content.py

@@ -1,9 +1,9 @@
 import ipaddress
 import json
-from aiohttp import web
 from base64 import b64decode
+
+from aiohttp import web
 from frozenlist import FrozenList
-from multidict import CIMultiDictProxy
 
 from pulpcore.plugin.content import app
 

pulp_service/pulp_service/app/middleware.py

@@ -3,22 +3,19 @@
 import logging
 import marshal
 import tempfile
-
 from contextlib import suppress
 from contextvars import ContextVar
-from os import getenv
 
 from django.contrib.auth import login
 from django.db import IntegrityError
 from django.utils.deprecation import MiddlewareMixin
 
-
-from pulpcore.metrics import init_otel_meter
 from pulpcore.app.util import get_worker_name
-from pulpcore.plugin.models import Artifact, Repository
-from pulpcore.plugin.util import extract_pk, get_artifact_url, resolve_prn
-from pulp_service.app.authentication import RHSamlAuthentication
+from pulpcore.metrics import init_otel_meter
+from pulpcore.plugin.models import Artifact
+from pulpcore.plugin.util import get_artifact_url
 
+from pulp_service.app.authentication import RHSamlAuthentication
 
 _logger = logging.getLogger(__name__)
 repository_name_var = ContextVar("repository_name")
@@ -66,7 +63,7 @@ def process_view(self, request, callback, callback_args, callback_kwargs):
             except Exception:
                 # we want the process_exception middleware to fire
                 # https://code.djangoproject.com/ticket/12250
-                return
+                return None
 
     def process_response(self, request, response):
         if hasattr(request, self.PROFILER_REQUEST_ATTR_NAME):
@@ -81,7 +78,7 @@ def process_response(self, request, response):
                 artifact = Artifact.init_and_validate(temp_file.name)
                 with suppress(IntegrityError):
                     artifact.save()
-                _logger.info(f"Profile data URL: {get_artifact_url(artifact)}")
+                _logger.info("Profile data URL: %s", get_artifact_url(artifact))
 
         return response
 
@@ -119,7 +116,7 @@ class RHSamlAuthHeaderMiddleware(MiddlewareMixin):
     def process_view(self, request, *args, **kwargs):
         if "/pulp-mgmt/" in request.path:
             if "HTTP_X_RH_IDENTITY" in request.META:
-                _logger.debug(f"{request.META['HTTP_X_RH_IDENTITY']}")
+                _logger.debug("%s", request.META["HTTP_X_RH_IDENTITY"])
 
                 # Authenticate user using RHSamlAuthentication backend
                 if not request.user.is_authenticated:
@@ -131,7 +128,7 @@ def process_view(self, request, *args, **kwargs):
                         request.session.modified = True
                         # Update request.user for the current request
                         request.user = user
-                        _logger.info(f"User {user.username} authenticated for pulp-mgmt")
+                        _logger.info("User %s authenticated for pulp-mgmt", user.username)
                     else:
                         _logger.warning("Failed to authenticate user from RH Identity header")
 

pulp_service/pulp_service/app/models.py

@@ -2,28 +2,22 @@
 import json
 import logging
 import ssl
-
-import aiohttp
-import jq
-
 from base64 import b64decode
 from binascii import Error as Base64DecodeError
 from datetime import datetime
-from hashlib import sha256
 from gettext import gettext as _
+from hashlib import sha256
 
+import aiohttp
+import jq
 from django.conf import settings
-from django.db import models
-
 from django.contrib.postgres.fields import ArrayField, HStoreField
-
-from pulpcore.plugin.models import BaseModel, Content, Domain, Group, RepositoryVersion
-from pulpcore.plugin.models import AutoAddObjPermsMixin
-from pulpcore.plugin.util import get_domain_pk
+from django.db import models
 
 from pulpcore.app.models import HeaderContentGuard
-
 from pulpcore.cache import Cache
+from pulpcore.plugin.models import AutoAddObjPermsMixin, BaseModel, Content, Domain, Group, RepositoryVersion
+from pulpcore.plugin.util import get_domain_pk
 
 _logger = logging.getLogger(__name__)
 
@@ -71,21 +65,19 @@ async def fetch_feature():
                     return await response.json()
 
         try:
-            _logger.info(f"[{datetime.now()}] Making a request to feature service API ...")
+            _logger.info("[%s] Making a request to feature service API ...", datetime.now())
             response = asyncio.run(fetch_feature())
-            _logger.info(f"[{datetime.now()}] Got a response from feature service API!")
+            _logger.info("[%s] Got a response from feature service API!", datetime.now())
         except aiohttp.ClientResponseError as err:
             if err.status == 400:
-                _logger.error(
-                    "Failed to request information for a user. BadRequest. URL: {}".format(err.request_info.url)
-                )
+                _logger.error("Failed to request information for a user. BadRequest. URL: %s", err.request_info.url)
 
             if err.status == 403:
                 _logger.error(
                     "Failed to request information for a user. Permission Denied. Verify if the certificate is still valid."
                 )
 
-            _logger.warn(_("Failed to fetch the Subscription feature information for a user."))
+            _logger.warning(_("Failed to fetch the Subscription feature information for a user."))
             raise PermissionError(_("Access denied."))
 
         features_available = {feature["name"] for feature in response["features"]}
@@ -95,7 +87,7 @@ def permit(self, request):
         try:
             header_content = request.headers[self.header_name]
         except KeyError:
-            _logger.error("Access not allowed. Header {header_name} not found.".format(header_name=self.header_name))
+            _logger.error("Access not allowed. Header %s not found.", self.header_name)
             raise PermissionError(_("Access denied."))
 
         try:
@@ -109,7 +101,7 @@ def permit(self, request):
             json_path = jq.compile(self.jq_filter)
 
             if settings.AUTHENTICATION_HEADER_DEBUG:
-                _logger.info("Authentication Header Debug enabled: {header_value}".format(header_value=header_value))
+                _logger.info("Authentication Header Debug enabled: %s", header_value)
 
             header_value = json_path.input_value(header_value).first()
 
@@ -136,11 +128,11 @@ def permit(self, request):
                 account_allowed = json.loads(account_allowed)
 
             if not account_allowed:
-                _logger.warn("Access not allowed - Features not available for the user.")
+                _logger.warning("Access not allowed - Features not available for the user.")
                 raise PermissionError(_("Access denied."))
 
-        except aiohttp.ClientResponseError as err:
-            _logger.warn("Access not allowed - Failed to check for features.")
+        except aiohttp.ClientResponseError:
+            _logger.warning("Access not allowed - Failed to check for features.")
             raise PermissionError(_("Access denied."))
 
         return

pulp_service/pulp_service/app/serializers.py

@@ -1,35 +1,35 @@
 import json
 import logging
-
 from gettext import gettext as _
-from jsonschema import validate, ValidationError
+
+from jsonschema import ValidationError, validate
 from rest_framework import serializers
 
+from pulpcore.app.models import Repository
+from pulpcore.plugin.models import PulpTemporaryFile
 from pulpcore.plugin.serializers import (
     ContentGuardSerializer,
     DetailRelatedField,
     GetOrCreateSerializerMixin,
+    IdentityField,
     ModelSerializer,
+    RepositoryVersionRelatedField,
     ValidateFieldsMixin,
 )
-from pulpcore.plugin.models import PulpTemporaryFile
-from pulpcore.plugin.serializers import IdentityField, RepositoryVersionRelatedField
-from pulpcore.app.models import Repository
 
+from pulp_service.app.constants import (
+    NPM_PACKAGE_LOCK_SCHEMA,
+    OSV_RH_ECOSYSTEM_CPES_LABEL,
+    OSV_RH_ECOSYSTEM_LABEL,
+    PYTHON_REPOSITORY_PULP_TYPE,
+)
 from pulp_service.app.models import (
     AgentScanReport,
     FeatureContentGuard,
     PyPIYankMonitor,
     VulnerabilityReport,
     YankedPackageReport,
 )
-from pulp_service.app.constants import (
-    NPM_PACKAGE_LOCK_SCHEMA,
-    OSV_RH_ECOSYSTEM_CPES_LABEL,
-    OSV_RH_ECOSYSTEM_LABEL,
-    PYTHON_REPOSITORY_PULP_TYPE,
-)
-
 
 _logger = logging.getLogger(__name__)
 

pulp_service/pulp_service/app/signals.py

@@ -1,15 +1,14 @@
 import logging
 
 from django.conf import settings
+from django.contrib.auth import get_user_model
 from django.db.models.signals import post_migrate, post_save
 from django.dispatch import receiver
-from django.contrib.auth import get_user_model
 
 from pulpcore.plugin.models import Domain
 
-from pulp_service.app.models import DomainOrg
 from pulp_service.app.authorization import group_var
-
+from pulp_service.app.models import DomainOrg
 
 _logger = logging.getLogger(__name__)
 
@@ -29,7 +28,7 @@ def log_new_user(sender, instance, created, **kwargs):
         from pulp_service.app.middleware import request_path_var
 
         request_path = request_path_var.get(None)
-        _logger.info(f"New user created: username={instance.username}, route={request_path or 'unknown'}")
+        _logger.info("New user created: username=%s, route=%s", instance.username, request_path or "unknown")
 
 
 @receiver(post_save, sender=Domain)