linting: Rewriting code for better linting compliance

Linting rewrite and validation assisted by Claude Code + Codex.

Signed-off-by: Denys Fedoryshchenko <denys.f@collabora.com>

Author: nuclearcat

.github/workflows/main.yml

@@ -111,4 +111,4 @@ jobs:
 
       - name: Run ruff lint check (E/W/F/I + complexity policy)
         run: |
-          /opt/venv/bin/ruff check --line-length 110 --select E,W,F,I,C90 --ignore E203,W503 api tests
+          /opt/venv/bin/ruff check --line-length 110 --select E,W,F,I,C901 --ignore E203 api tests

api/admin.py

@@ -9,11 +9,11 @@
 
 """Command line utility for creating an admin user"""
 
-import asyncio
 import argparse
-import sys
+import asyncio
 import getpass
 import os
+import sys
 import pymongo
 
 from .auth import Authentication
@@ -42,13 +42,15 @@ async def setup_admin_user(db, username, email, password=None):
     hashed_password = Authentication.get_password_hash(password)
     print(f"Creating {username} user...")
     try:
-        return await db.create(User(
-            username=username,
-            hashed_password=hashed_password,
-            email=email,
-            is_superuser=1,
-            is_verified=1,
-        ))
+        return await db.create(
+            User(
+                username=username,
+                hashed_password=hashed_password,
+                email=email,
+                is_superuser=1,
+                is_verified=1,
+            )
+        )
     except pymongo.errors.DuplicateKeyError as exc:
         err = str(exc)
         if "username" in err:
@@ -73,19 +75,19 @@ async def main(args):
     return created is not None
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     parser = argparse.ArgumentParser("Create KernelCI API admin user")
-    parser.add_argument('--mongo', default='mongodb://db:27017',
-                        help="Mongo server connection string")
-    parser.add_argument('--username', default='admin',
-                        help="Admin username")
-    parser.add_argument('--database', default='kernelci',
-                        help="KernelCI database name")
-    parser.add_argument('--email', required=True,
-                        help="Admin user email address")
     parser.add_argument(
-        '--password',
-        default='',
+        "--mongo",
+        default="mongodb://db:27017",
+        help="Mongo server connection string",
+    )
+    parser.add_argument("--username", default="admin", help="Admin username")
+    parser.add_argument("--database", default="kernelci", help="KernelCI database name")
+    parser.add_argument("--email", required=True, help="Admin user email address")
+    parser.add_argument(
+        "--password",
+        default="",
         help="Admin password (if empty, falls back to KCI_INITIAL_PASSWORD)",
     )
     arguments = parser.parse_args()

api/auth.py

@@ -6,12 +6,13 @@
 
 """User authentication utilities"""
 
-from passlib.context import CryptContext
 from fastapi_users.authentication import (
     AuthenticationBackend,
     BearerTransport,
     JWTStrategy,
 )
+from passlib.context import CryptContext
+
 from .config import AuthSettings
 
 
@@ -34,7 +35,7 @@ def get_jwt_strategy(self) -> JWTStrategy:
         return JWTStrategy(
             secret=self._settings.secret_key,
             algorithm=self._settings.algorithm,
-            lifetime_seconds=self._settings.access_token_expire_seconds
+            lifetime_seconds=self._settings.access_token_expire_seconds,
         )
 
     def get_user_authentication_backend(self):

api/config.py

@@ -12,6 +12,7 @@
 # pylint: disable=too-few-public-methods
 class AuthSettings(BaseSettings):
     """Authentication settings"""
+
     secret_key: str
     algorithm: str = "HS256"
     # Set to None so tokens don't expire
@@ -23,6 +24,7 @@ class AuthSettings(BaseSettings):
 # pylint: disable=too-few-public-methods
 class PubSubSettings(BaseSettings):
     """Pub/Sub settings loaded from the environment"""
+
     cloud_events_source: str = "https://api.kernelci.org/"
     redis_host: str = "redis"
     redis_db_number: int = 1
@@ -36,6 +38,7 @@ class PubSubSettings(BaseSettings):
 # pylint: disable=too-few-public-methods
 class EmailSettings(BaseSettings):
     """Email settings"""
+
     smtp_host: str
     smtp_port: int
     email_sender: EmailStr

api/db.py

@@ -6,14 +6,19 @@
 
 """Database abstraction"""
 
-from bson import ObjectId
 from beanie import init_beanie
+from bson import ObjectId
 from fastapi_pagination.ext.motor import paginate
-from motor import motor_asyncio
-from redis import asyncio as aioredis
 from kernelci.api.models import (
-    EventHistory, Hierarchy, Node, TelemetryEvent, parse_node_obj
+    EventHistory,
+    Hierarchy,
+    Node,
+    TelemetryEvent,
+    parse_node_obj,
 )
+from motor import motor_asyncio
+from redis import asyncio as aioredis
+
 from .models import User, UserGroup
 
 
@@ -26,33 +31,30 @@ class Database:
     """
 
     COLLECTIONS = {
-        User: 'user',
-        Node: 'node',
-        UserGroup: 'usergroup',
-        EventHistory: 'eventhistory',
-        TelemetryEvent: 'telemetry',
+        User: "user",
+        Node: "node",
+        UserGroup: "usergroup",
+        EventHistory: "eventhistory",
+        TelemetryEvent: "telemetry",
     }
 
     OPERATOR_MAP = {
-        'lt': '$lt',
-        'lte': '$lte',
-        'gt': '$gt',
-        'gte': '$gte',
-        'ne': '$ne',
-        're': '$regex',
-        'in': '$in',
-        'nin': '$nin',
+        "lt": "$lt",
+        "lte": "$lte",
+        "gt": "$gt",
+        "gte": "$gte",
+        "ne": "$ne",
+        "re": "$regex",
+        "in": "$in",
+        "nin": "$nin",
     }
 
-    BOOL_VALUE_MAP = {
-        'true': True,
-        'false': False
-    }
+    BOOL_VALUE_MAP = {"true": True, "false": False}
 
-    def __init__(self, service='mongodb://db:27017', db_name='kernelci'):
+    def __init__(self, service="mongodb://db:27017", db_name="kernelci"):
         self._motor = motor_asyncio.AsyncIOMotorClient(service)
         # TBD: Make redis host configurable
-        self._redis = aioredis.from_url('redis://redis:6379')
+        self._redis = aioredis.from_url("redis://redis:6379")
         self._db = self._motor[db_name]
 
     async def initialize_beanie(self):
@@ -143,14 +145,13 @@ def _translate_operators(self, attributes):
                 for op_name, op_value in value.items():
                     op_key = self.OPERATOR_MAP.get(op_name)
                     if op_key:
-                        if op_key in ('$in', '$nin'):
+                        if op_key in ("$in", "$nin"):
                             # Create a list of values from ',' separated string
                             op_value = op_value.split(",")
                         if isinstance(op_value, str) and op_value.isdecimal():
                             op_value = int(op_value)
                         if translated_attributes.get(key):
-                            translated_attributes[key].update({
-                                op_key: op_value})
+                            translated_attributes[key].update({op_key: op_value})
                         else:
                             translated_attributes[key] = {op_key: op_value}
         return translated_attributes
@@ -160,7 +161,7 @@ def _convert_int_values(cls, attributes):
         for key, val in attributes.items():
             if isinstance(val, dict):
                 for sub_key, sub_val in val.items():
-                    if sub_key == 'int':
+                    if sub_key == "int":
                         attributes[key] = int(sub_val)
         return attributes
 
@@ -205,14 +206,13 @@ async def find_by_attributes_nonpaginated(self, model, attributes):
         query = self._prepare_query(attributes)
         # find "limit" and "offset" keys in the query, retrieve them and
         # remove them from the query
-        limit = query.pop('limit', None)
-        offset = query.pop('offset', None)
+        limit = query.pop("limit", None)
+        offset = query.pop("offset", None)
         # convert to int if limit and offset are strings
         limit = int(limit) if limit is not None else None
         offset = int(offset) if offset is not None else None
         if limit is not None and offset is not None:
-            return await (col.find(query)
-                          .skip(offset).limit(limit).to_list(None))
+            return await col.find(query).skip(offset).limit(limit).to_list(None)
         if limit is not None:
             return await col.find(query).limit(limit).to_list(None)
         if offset is not None:
@@ -239,7 +239,7 @@ async def create(self, obj):
         """
         if obj.id is not None:
             raise ValueError(f"Object cannot be created with id: {obj.id}")
-        delattr(obj, 'id')
+        delattr(obj, "id")
         col = self._get_collection(obj.__class__)
         res = await col.insert_one(obj.model_dump(by_alias=True))
         obj.id = res.inserted_id
@@ -251,22 +251,19 @@ async def insert_many(self, model, documents):
         result = await col.insert_many(documents)
         return result.inserted_ids
 
-    async def _create_recursively(self, hierarchy: Hierarchy, parent: Node,
-                                  cls, col):
+    async def _create_recursively(self, hierarchy: Hierarchy, parent: Node, cls, col):
         obj = parse_node_obj(hierarchy.node)
         if parent:
             obj.parent = parent.id
         if obj.id:
             obj.update()
             if obj.parent == obj.id:
                 raise ValueError("Parent cannot be the same as the object")
-            res = await col.replace_one(
-                {'_id': ObjectId(obj.id)}, obj.dict(by_alias=True)
-            )
+            res = await col.replace_one({"_id": ObjectId(obj.id)}, obj.dict(by_alias=True))
             if res.matched_count == 0:
                 raise ValueError(f"No object found with id: {obj.id}")
         else:
-            delattr(obj, 'id')
+            delattr(obj, "id")
             res = await col.insert_one(obj.dict(by_alias=True))
             obj.id = res.inserted_id
         obj = cls(**await col.find_one(ObjectId(obj.id)))
@@ -296,9 +293,7 @@ async def update(self, obj):
             obj.update()
             if obj.parent == obj.id:
                 raise ValueError("Parent cannot be the same as the object")
-        res = await col.replace_one(
-            {'_id': ObjectId(obj.id)}, obj.dict(by_alias=True)
-        )
+        res = await col.replace_one({"_id": ObjectId(obj.id)}, obj.dict(by_alias=True))
         if res.matched_count == 0:
             raise ValueError(f"No object found with id: {obj.id}")
         return obj.__class__(**await col.find_one(ObjectId(obj.id)))

api/email_sender.py

@@ -7,43 +7,43 @@
 
 """SMTP Email Sender module"""
 
-from email.mime.multipart import MIMEMultipart
 import email
 import email.mime.text
 import smtplib
+from email.mime.multipart import MIMEMultipart
+
 from fastapi import HTTPException, status
+
 from .config import EmailSettings
 
 
 class EmailSender:  # pylint: disable=too-few-public-methods
     """Class to send email report using SMTP"""
+
     def __init__(self):
         self._settings = EmailSettings()
 
     def _smtp_connect(self):
         """Method to create a connection with SMTP server"""
         if self._settings.smtp_port == 465:
-            smtp = smtplib.SMTP_SSL(self._settings.smtp_host,
-                                    self._settings.smtp_port)
+            smtp = smtplib.SMTP_SSL(self._settings.smtp_host, self._settings.smtp_port)
         else:
-            smtp = smtplib.SMTP(self._settings.smtp_host,
-                                self._settings.smtp_port)
+            smtp = smtplib.SMTP(self._settings.smtp_host, self._settings.smtp_port)
             smtp.starttls()
-        smtp.login(self._settings.email_sender,
-                   self._settings.email_password)
+        smtp.login(self._settings.email_sender, self._settings.email_password)
         return smtp
 
     def _create_email(self, email_subject, email_content, email_recipient):
         """Method to create an email message from email subject, contect,
         sender, and receiver"""
         email_msg = MIMEMultipart()
         email_text = email.mime.text.MIMEText(email_content, "plain", "utf-8")
-        email_text.replace_header('Content-Transfer-Encoding', 'quopri')
-        email_text.set_payload(email_content, 'utf-8')
+        email_text.replace_header("Content-Transfer-Encoding", "quopri")
+        email_text.set_payload(email_content, "utf-8")
         email_msg.attach(email_text)
-        email_msg['To'] = email_recipient
-        email_msg['From'] = self._settings.email_sender
-        email_msg['Subject'] = email_subject
+        email_msg["To"] = email_recipient
+        email_msg["From"] = self._settings.email_sender
+        email_msg["Subject"] = email_subject
         return email_msg
 
     def _send_email(self, email_msg):
@@ -57,13 +57,10 @@ def _send_email(self, email_msg):
             print(f"Error in sending email: {str(exc)}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-                detail="Failed to send email"
-            )from exc
+                detail="Failed to send email",
+            ) from exc
 
-    def create_and_send_email(self, email_subject, email_content,
-                              email_recipient):
+    def create_and_send_email(self, email_subject, email_content, email_recipient):
         """Method to create and send email"""
-        email_msg = self._create_email(
-            email_subject, email_content, email_recipient
-        )
+        email_msg = self._create_email(email_subject, email_content, email_recipient)
         self._send_email(email_msg)