lint con ruff

src/ftp_csv_receiver.py

@@ -1,22 +1,21 @@
 #!.venv/bin/python
 """
-    This module implements an FTP server with custom commands for 
-    managing virtual users and handling CSV file uploads.
+This module implements an FTP server with custom commands for
+managing virtual users and handling CSV file uploads.
 """
 
-import os
 import logging
-
+import os
 from hashlib import sha256
 from pathlib import Path
 
+from pyftpdlib.authorizers import AuthenticationFailed, DummyAuthorizer
 from pyftpdlib.handlers import FTPHandler
 from pyftpdlib.servers import FTPServer
-from pyftpdlib.authorizers import DummyAuthorizer, AuthenticationFailed
 
 from utils.config import loader_ftp_csv as setting
+from utils.connect import file_management, user_admin
 from utils.database.connection import connetti_db
-from utils.connect import user_admin, file_management
 
 # Configure logging (moved inside main function)
 
@@ -33,46 +32,37 @@ class DummySha256Authorizer(DummyAuthorizer):
             cfg: The configuration object.
         """
         super().__init__()
-        self.add_user(
-            cfg.adminuser[0], cfg.adminuser[1], cfg.adminuser[2], perm=cfg.adminuser[3]
-        )
+        self.add_user(cfg.adminuser[0], cfg.adminuser[1], cfg.adminuser[2], perm=cfg.adminuser[3])
 
         # Define the database connection
         conn = connetti_db(cfg)
 
         # Create a cursor
         cur = conn.cursor()
-        cur.execute(
-            f"SELECT ftpuser, hash, virtpath, perm FROM {cfg.dbname}.{cfg.dbusertable} WHERE disabled_at IS NULL"
-        )
+        cur.execute(f"SELECT ftpuser, hash, virtpath, perm FROM {cfg.dbname}.{cfg.dbusertable} WHERE disabled_at IS NULL")
 
         for ftpuser, user_hash, virtpath, perm in cur.fetchall():
             # Create the user's directory if it does not exist.
             try:
                 Path(cfg.virtpath + ftpuser).mkdir(parents=True, exist_ok=True)
                 self.add_user(ftpuser, user_hash, virtpath, perm)
-            except Exception as e: # pylint: disable=broad-except
+            except Exception as e:  # pylint: disable=broad-except
                 self.responde(f"551 Error in create virtual user path: {e}")
-            
 
-    def validate_authentication(
-        self: object, username: str, password: str, handler: object
-    ) -> None:
+    def validate_authentication(self: object, username: str, password: str, handler: object) -> None:
         # Validate the user's password against the stored user_hash
         user_hash = sha256(password.encode("UTF-8")).hexdigest()
         try:
             if self.user_table[username]["pwd"] != user_hash:
                 raise KeyError
         except KeyError:
-            raise AuthenticationFailed
+            raise AuthenticationFailed  # noqa: B904
 
 
 class ASEHandler(FTPHandler):
     """Custom FTP handler that extends FTPHandler with custom commands and file handling."""
 
-    def __init__(
-        self: object, conn: object, server: object, ioloop: object = None
-    ) -> None:
+    def __init__(self: object, conn: object, server: object, ioloop: object = None) -> None:
         """Initializes the handler, adds custom commands, and sets up command permissions.
 
         Args:
@@ -85,42 +75,42 @@ class ASEHandler(FTPHandler):
         # Add custom FTP commands for managing virtual users - command in lowercase
         self.proto_cmds.update(
             {
-                "SITE ADDU": dict(
-                    perm="M",
-                    auth=True,
-                    arg=True,
-                    help="Syntax: SITE <SP> ADDU USERNAME PASSWORD (add virtual user).",
-                )
+                "SITE ADDU": {
+                    "perm": "M",
+                    "auth": True,
+                    "arg": True,
+                    "help": "Syntax: SITE <SP> ADDU USERNAME PASSWORD (add virtual user).",
+                }
             }
         )
         self.proto_cmds.update(
             {
-                "SITE DISU": dict(
-                    perm="M",
-                    auth=True,
-                    arg=True,
-                    help="Syntax: SITE <SP> DISU USERNAME (disable virtual user).",
-                )
+                "SITE DISU": {
+                    "perm": "M",
+                    "auth": True,
+                    "arg": True,
+                    "help": "Syntax: SITE <SP> DISU USERNAME (disable virtual user).",
+                }
             }
         )
         self.proto_cmds.update(
             {
-                "SITE ENAU": dict(
-                    perm="M",
-                    auth=True,
-                    arg=True,
-                    help="Syntax: SITE <SP> ENAU USERNAME (enable virtual user).",
-                )
+                "SITE ENAU": {
+                    "perm": "M",
+                    "auth": True,
+                    "arg": True,
+                    "help": "Syntax: SITE <SP> ENAU USERNAME (enable virtual user).",
+                }
             }
         )
         self.proto_cmds.update(
             {
-                "SITE LSTU": dict(
-                    perm="M",
-                    auth=True,
-                    arg=None,
-                    help="Syntax: SITE <SP> LSTU (list virtual users).",
-                )
+                "SITE LSTU": {
+                    "perm": "M",
+                    "auth": True,
+                    "arg": None,
+                    "help": "Syntax: SITE <SP> LSTU (list virtual users).",
+                }
             }
         )