From 396cf83a0599f001b8e72ec7583d8a5bf47f2551 Mon Sep 17 00:00:00 2001 From: Brian Bjarke Jensen Date: Wed, 17 Sep 2025 12:14:37 +0200 Subject: [PATCH] code quality fixes --- python_repositories/adapters/minio_adapter.py | 23 ++++++++----------- python_repositories/adapters/redis_adapter.py | 19 ++++++++------- 2 files changed, 19 insertions(+), 23 deletions(-) diff --git a/python_repositories/adapters/minio_adapter.py b/python_repositories/adapters/minio_adapter.py index 241cfeb..965ec4f 100644 --- a/python_repositories/adapters/minio_adapter.py +++ b/python_repositories/adapters/minio_adapter.py @@ -3,7 +3,7 @@ from __future__ import annotations import os from io import BytesIO -from typing import TYPE_CHECKING +from importlib.util import find_spec import structlog from python_utils import check_env @@ -12,9 +12,11 @@ from python_repositories.interfaces import ( ConnectionAwareInterface, ) -# Import for mypy type checking only -if TYPE_CHECKING: - from minio import Minio, S3Error +# Handle optional dependencies +if find_spec("minio") is not None: + import minio +else: + minio = None class MinioAdapter( @@ -30,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 - except ImportError as e: - raise RuntimeError("MinioAdapter dependencies missing") from e # Setup logger self.logger = structlog.get_logger( self.__class__.__name__, @@ -49,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: @@ -84,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, @@ -114,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 @@ -168,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}'" diff --git a/python_repositories/adapters/redis_adapter.py b/python_repositories/adapters/redis_adapter.py index 9f40db9..e8d845b 100644 --- a/python_repositories/adapters/redis_adapter.py +++ b/python_repositories/adapters/redis_adapter.py @@ -1,7 +1,8 @@ """Definition of RedisAdapter class.""" from __future__ import annotations -from typing import cast, TYPE_CHECKING +from typing import cast +from importlib.util import find_spec import os import structlog @@ -12,10 +13,13 @@ from python_repositories.interfaces import ( ConnectionAwareInterface, ) -# Import for mypy type checking only -if TYPE_CHECKING: +# 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( @@ -25,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 - from redis.commands.json.path import Path as RedisPath - except ImportError as e: - raise RuntimeError("RedisAdapter dependencies missing") from e # Setup logger self.logger = structlog.get_logger( self.__class__.__name__, @@ -43,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."""