Refactor: modernize codebase by replacing deprecated patterns, simplifying imports, and applying type hinting refinements.

src/easymapping/__init__.py

@@ -4,6 +4,7 @@ import os
 import re
 
 from jinja2 import Environment, FileSystemLoader
+
 from functions import loggerEasyHaproxy
 
 
@@ -17,7 +18,7 @@ class DockerLabelHandler:
 
     def create(self, key):
         if isinstance(key, str):
-            return "{}.{}".format(self.__label_base, key)
+            return f"{self.__label_base}.{key}"
 
         return "{}.{}".format(self.__label_base, ".".join(key))
 
@@ -140,7 +141,7 @@ class HaproxyConfigGenerator:
 
             self.label.set_data(d)
 
-            # Parse each definition found. 
+            # Parse each definition found.
             for definition in sorted(definitions.keys()):
                 mode = self.label.get(
                     self.label.create([definition, "mode"]),
@@ -209,7 +210,7 @@ class HaproxyConfigGenerator:
                     if socket_path:
                         server_address = socket_path
                     else:
-                        server_address = "{}:{}".format(container, ct_port)
+                        server_address = f"{container}:{ct_port}"
 
                     easymapping[port]["hosts"][hostname]["containers"] += [server_address]
                     easymapping[port]["hosts"][hostname]["certbot"] = certbot
@@ -306,7 +307,7 @@ class HaproxyConfigGenerator:
                     # handle SSL
                     ssl_label = self.label.create([definition, "sslcert"])
                     if self.label.has_label(ssl_label):
-                        filename = "{}.pem".format(d[host_label])
+                        filename = f"{d[host_label]}.pem"
                         easymapping[port]["ssl"] = True if not clone_to_ssl else False
                         self.certs[filename] = base64.b64decode(d[ssl_label]).decode('ascii')
 

src/functions/__init__.py

@@ -1,16 +1,18 @@
+import logging
 import os
 import shlex
 import subprocess
 import sys
-import psutil
 import time
-import logging
 from datetime import datetime
 from multiprocessing import Process
 from typing import Final
+
+import psutil
 import requests
 from OpenSSL import crypto
 
+
 class ContainerEnv:
     @staticmethod
     def read():
@@ -150,7 +152,7 @@ class Functions:
 
     @staticmethod
     def load(filename):
-        with open(filename, 'r') as content_file:
+        with open(filename) as content_file:
             return content_file.read()
 
     @staticmethod
@@ -384,7 +386,7 @@ class Certbot:
                                     )
 
             if self.certbot_manual_auth_hook:
-                certbot_certonly += '    --manual --manual-auth-hook \'{hook}\''.format(hook=self.certbot_manual_auth_hook)
+                certbot_certonly += f'    --manual --manual-auth-hook \'{self.certbot_manual_auth_hook}\''
 
             if loggerCertbot.level == logging.DEBUG:
                 certbot_certonly += '    -v'

src/main.py

@@ -2,8 +2,14 @@ import os
 
 from deepdiff import DeepDiff
 
-from functions import Functions, DaemonizeHAProxy, Certbot, Consts, loggerInit, loggerEasyHaproxy, loggerHaproxy, \
-    loggerCertbot
+from functions import (
+    Certbot,
+    Consts,
+    DaemonizeHAProxy,
+    Functions,
+    loggerEasyHaproxy,
+    loggerInit,
+)
 from processor import ProcessorInterface
 
 
@@ -69,7 +75,7 @@ def main():
     loggerInit.debug('Environment:')
     for name, value in os.environ.items():
         if "HAPROXY" in name:
-            loggerInit.debug("- {0}: {1}".format(name, value))
+            loggerInit.debug(f"- {name}: {value}")
 
     start()
 

src/plugins/__init__.py

@@ -1,10 +1,11 @@
-import os
 import importlib.util
+import os
 import sys
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Optional, Dict, Any, List
+from typing import Any, Dict, List, Optional
+
 from functions import loggerEasyHaproxy
 
 
@@ -20,17 +21,17 @@ class PluginContext:
     parsed_object: dict              # {IP: labels} from discovery
     easymapping: list               # Current HAProxy mapping structure
     container_env: dict             # Environment configuration
-    domain: Optional[str] = None    # Domain name (for DOMAIN plugins)
-    port: Optional[str] = None      # Port (for DOMAIN plugins)
-    host_config: Optional[dict] = None  # Domain-specific config
+    domain: str | None = None    # Domain name (for DOMAIN plugins)
+    port: str | None = None      # Port (for DOMAIN plugins)
+    host_config: dict | None = None  # Domain-specific config
 
 
 @dataclass
 class PluginResult:
     """Plugin execution result"""
     haproxy_config: str = ""                     # HAProxy config snippet to inject
-    modified_easymapping: Optional[list] = None  # Modified easymapping structure
-    metadata: Dict[str, Any] = field(default_factory=dict)  # Plugin metadata for logging
+    modified_easymapping: list | None = None  # Modified easymapping structure
+    metadata: dict[str, Any] = field(default_factory=dict)  # Plugin metadata for logging
 
 
 class PluginInterface(ABC):
@@ -85,9 +86,9 @@ class PluginManager:
         """
         self.plugins_dir = plugins_dir
         self.abort_on_error = abort_on_error
-        self.plugins: Dict[str, PluginInterface] = {}
-        self.global_plugins: List[PluginInterface] = []
-        self.domain_plugins: List[PluginInterface] = []
+        self.plugins: dict[str, PluginInterface] = {}
+        self.global_plugins: list[PluginInterface] = []
+        self.domain_plugins: list[PluginInterface] = []
         self.logger = loggerEasyHaproxy
 
     def load_plugins(self) -> None:
@@ -174,7 +175,7 @@ class PluginManager:
             except Exception as e:
                 self._handle_error(f"Failed to configure plugin '{plugin_name}': {str(e)}")
 
-    def execute_global_plugins(self, context: PluginContext, enabled_list: Optional[List[str]] = None) -> List[PluginResult]:
+    def execute_global_plugins(self, context: PluginContext, enabled_list: list[str] | None = None) -> list[PluginResult]:
         """
         Execute all global plugins
 
@@ -205,7 +206,7 @@ class PluginManager:
 
         return results
 
-    def execute_domain_plugins(self, context: PluginContext, enabled_list: Optional[List[str]] = None) -> List[PluginResult]:
+    def execute_domain_plugins(self, context: PluginContext, enabled_list: list[str] | None = None) -> list[PluginResult]:
         """
         Execute all domain plugins for a specific domain
 

src/plugins/builtin/cleanup.py

@@ -21,16 +21,16 @@ Example Environment Variable:
     EASYHAPROXY_PLUGIN_CLEANUP_MAX_IDLE_TIME=600
 """
 
+import glob
 import os
 import sys
-import glob
 import time
 
 # Add parent directory to path for imports
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from plugins import PluginInterface, PluginType, PluginContext, PluginResult
 from functions import loggerEasyHaproxy
+from plugins import PluginContext, PluginInterface, PluginResult, PluginType
 
 
 class CleanupPlugin(PluginInterface):

src/plugins/builtin/cloudflare.py

@@ -33,8 +33,8 @@ import sys
 # Add parent directory to path for imports
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from plugins import PluginInterface, PluginType, PluginContext, PluginResult
 from functions import loggerEasyHaproxy
+from plugins import PluginContext, PluginInterface, PluginResult, PluginType
 
 
 class CloudflarePlugin(PluginInterface):

src/plugins/builtin/deny_pages.py

@@ -32,7 +32,7 @@ import sys
 # Add parent directory to path for imports
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from plugins import PluginInterface, PluginType, PluginContext, PluginResult
+from plugins import PluginContext, PluginInterface, PluginResult, PluginType
 
 
 class DenyPagesPlugin(PluginInterface):

src/plugins/builtin/fastcgi.py

@@ -42,8 +42,7 @@ import sys
 # Add parent directory to path for imports
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from plugins import PluginInterface, PluginType, PluginContext, PluginResult
-from functions import loggerEasyHaproxy
+from plugins import PluginContext, PluginInterface, PluginResult, PluginType
 
 
 class FastcgiPlugin(PluginInterface):
@@ -124,7 +123,7 @@ class FastcgiPlugin(PluginInterface):
 
         # PATH_INFO support
         if self.path_info:
-            fcgi_app_lines.append(f"    path-info ^(/.+\\.php)(/.*)?$")
+            fcgi_app_lines.append("    path-info ^(/.+\\.php)(/.*)?$")
 
         # Set SCRIPT_FILENAME if customized
         if self.script_filename and self.script_filename != "%[path]":

src/plugins/builtin/ip_whitelist.py

@@ -33,7 +33,7 @@ import sys
 # Add parent directory to path for imports
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from plugins import PluginInterface, PluginType, PluginContext, PluginResult
+from plugins import PluginContext, PluginInterface, PluginResult, PluginType
 
 
 class IpWhitelistPlugin(PluginInterface):

src/plugins/builtin/jwt_validator.py

@@ -74,8 +74,8 @@ import sys
 # Add parent directory to path for imports
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from plugins import PluginInterface, PluginType, PluginContext, PluginResult
 from functions import loggerEasyHaproxy
+from plugins import PluginContext, PluginInterface, PluginResult, PluginType
 
 
 class JwtValidatorPlugin(PluginInterface):

src/processor/__init__.py

@@ -8,8 +8,7 @@ from kubernetes import client, config
 from kubernetes.client.rest import ApiException
 
 from easymapping import HaproxyConfigGenerator
-from functions import Functions, Consts, ContainerEnv
-from functions import loggerEasyHaproxy
+from functions import Consts, ContainerEnv, Functions, loggerEasyHaproxy
 
 
 class ProcessorInterface:
@@ -87,7 +86,7 @@ class ProcessorInterface:
 
     def save_certs(self, path):
         for cert in self.get_certs():
-            Functions.save("{0}/{1}".format(path, cert), self.get_certs(cert))
+            Functions.save(f"{path}/{cert}", self.get_certs(cert))
 
 
 class Static(ProcessorInterface):
@@ -502,7 +501,7 @@ class Kubernetes(ProcessorInterface):
                         if tls.secret_name not in self.cert_cache or self.cert_cache[tls.secret_name] != secret.data:
                             self.cert_cache[tls.secret_name] = secret.data
                             Functions.save(
-                                "{0}/{1}.pem".format(Consts.certs_haproxy, tls.secret_name),
+                                f"{Consts.certs_haproxy}/{tls.secret_name}.pem",
                                 base64.b64decode(secret.data["tls.crt"]).decode('ascii') + "\n" + base64.b64decode(
                                     secret.data["tls.key"]).decode('ascii')
                             )