code quality fixes

python_repositories/adapters/minio_adapter.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 import os
 from io import BytesIO
+from importlib.util import find_spec
 import structlog
 from python_utils import check_env
 
@@ -11,6 +12,12 @@ from python_repositories.interfaces import (
     ConnectionAwareInterface,
 )
 
+# Handle optional dependencies
+if find_spec("minio") is not None:
+    import minio
+else:
+    minio = None
+
 
 class MinioAdapter(
     ContextAwareInterface,
@@ -25,11 +32,6 @@ class MinioAdapter(
     chunk_size: int = 5 * 2**20  # 5 MiB
 
     def __init__(self) -> None:
-        # Import here to avoid hard dependency if MinioAdapter is not used
-        try:
-            from minio import Minio, S3Error  # type: ignore[import]
-        except ImportError as e:
-            raise RuntimeError("MinioAdapter dependencies missing") from e
         # Setup logger
         self.logger = structlog.get_logger(
             self.__class__.__name__,
@@ -44,7 +46,7 @@ class MinioAdapter(
             },
         )
         # Prepare internal variables
-        self._client: Minio | None = None
+        self._client: minio.Minio | None = None
         self._bucket_name: str | None = None
 
     def __enter__(self) -> MinioAdapter:
@@ -79,7 +81,7 @@ class MinioAdapter(
         secret_key = str(os.getenv(self.secret_key_env_var_name))
         bucket = str(os.getenv(self.bucket_env_var_name))
         # Connect client
-        client = Minio(
+        client = minio.Minio(
             endpoint=endpoint,
             access_key=access_key,
             secret_key=secret_key,
@@ -109,7 +111,7 @@ class MinioAdapter(
     @property
     def is_connected(self) -> bool:
         """Check if connected to Minio server."""
-        res = bool(isinstance(self._client, Minio))
+        res = bool(isinstance(self._client, minio.Minio))
         self.logger.debug(res)
         return res
 
@@ -163,7 +165,7 @@ class MinioAdapter(
                 f"Got object '{object_name}' from bucket '{self._bucket_name}'"
             )
             return buffer
-        except S3Error as exc:
+        except minio.S3Error as exc:
             if exc.code == "NoSuchKey":
                 self.logger.warning(
                     f"Object '{object_name}' not found in bucket '{self._bucket_name}'"

python_repositories/adapters/redis_adapter.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 from typing import cast
+from importlib.util import find_spec
 import os
 import structlog
 
@@ -12,6 +13,14 @@ from python_repositories.interfaces import (
     ConnectionAwareInterface,
 )
 
+# Handle optional dependencies
+if find_spec("redis") is not None:
+    import redis
+    from redis.commands.json.path import Path as RedisPath
+else:
+    redis = None  # type: ignore
+    RedisPath = None  # type: ignore
+
 
 class RedisAdapter(
     ContextAwareInterface,
@@ -20,16 +29,10 @@ class RedisAdapter(
     """Redis adapter exposing basic CRUD functionality."""
 
     uri_env_var_name: str = "REDIS_URI"
-    path: str = RedisPath.root_path()
+    path: str = "."  # JSON root path, updated in __init__
     encoding: str = "UTF-8"
 
     def __init__(self) -> None:
-        # Import here to avoid hard dependency to optional package
-        try:
-            import redis  # type: ignore[import]
-            from redis.commands.json.path import Path as RedisPath  # type: ignore[import]
-        except ImportError as e:
-            raise RuntimeError("RedisAdapter dependencies missing") from e
         # Setup logger
         self.logger = structlog.get_logger(
             self.__class__.__name__,
@@ -38,6 +41,7 @@ class RedisAdapter(
         check_env(self.uri_env_var_name)
         # Prepare internal variables
         self._client: redis.Redis | None = None
+        self.path: str = RedisPath.root_path()
 
     def __enter__(self) -> RedisAdapter:
         """Enter the context."""