code quality fixes
Code Quality Pipeline / code-quality (pull_request) Successful in 30s
Test Python Package / test (pull_request) Successful in 1m12s

This commit is contained in:
Brian Bjarke Jensen
2025-09-17 12:14:37 +02:00
parent 538b26b62f
commit 396cf83a05
2 changed files with 19 additions and 23 deletions
+10 -13
View File
@@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
import os import os
from io import BytesIO from io import BytesIO
from typing import TYPE_CHECKING from importlib.util import find_spec
import structlog import structlog
from python_utils import check_env from python_utils import check_env
@@ -12,9 +12,11 @@ from python_repositories.interfaces import (
ConnectionAwareInterface, ConnectionAwareInterface,
) )
# Import for mypy type checking only # Handle optional dependencies
if TYPE_CHECKING: if find_spec("minio") is not None:
from minio import Minio, S3Error import minio
else:
minio = None
class MinioAdapter( class MinioAdapter(
@@ -30,11 +32,6 @@ class MinioAdapter(
chunk_size: int = 5 * 2**20 # 5 MiB chunk_size: int = 5 * 2**20 # 5 MiB
def __init__(self) -> None: 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 # Setup logger
self.logger = structlog.get_logger( self.logger = structlog.get_logger(
self.__class__.__name__, self.__class__.__name__,
@@ -49,7 +46,7 @@ class MinioAdapter(
}, },
) )
# Prepare internal variables # Prepare internal variables
self._client: Minio | None = None self._client: minio.Minio | None = None
self._bucket_name: str | None = None self._bucket_name: str | None = None
def __enter__(self) -> MinioAdapter: def __enter__(self) -> MinioAdapter:
@@ -84,7 +81,7 @@ class MinioAdapter(
secret_key = str(os.getenv(self.secret_key_env_var_name)) secret_key = str(os.getenv(self.secret_key_env_var_name))
bucket = str(os.getenv(self.bucket_env_var_name)) bucket = str(os.getenv(self.bucket_env_var_name))
# Connect client # Connect client
client = Minio( client = minio.Minio(
endpoint=endpoint, endpoint=endpoint,
access_key=access_key, access_key=access_key,
secret_key=secret_key, secret_key=secret_key,
@@ -114,7 +111,7 @@ class MinioAdapter(
@property @property
def is_connected(self) -> bool: def is_connected(self) -> bool:
"""Check if connected to Minio server.""" """Check if connected to Minio server."""
res = bool(isinstance(self._client, Minio)) res = bool(isinstance(self._client, minio.Minio))
self.logger.debug(res) self.logger.debug(res)
return res return res
@@ -168,7 +165,7 @@ class MinioAdapter(
f"Got object '{object_name}' from bucket '{self._bucket_name}'" f"Got object '{object_name}' from bucket '{self._bucket_name}'"
) )
return buffer return buffer
except S3Error as exc: except minio.S3Error as exc:
if exc.code == "NoSuchKey": if exc.code == "NoSuchKey":
self.logger.warning( self.logger.warning(
f"Object '{object_name}' not found in bucket '{self._bucket_name}'" f"Object '{object_name}' not found in bucket '{self._bucket_name}'"
+9 -10
View File
@@ -1,7 +1,8 @@
"""Definition of RedisAdapter class.""" """Definition of RedisAdapter class."""
from __future__ import annotations from __future__ import annotations
from typing import cast, TYPE_CHECKING from typing import cast
from importlib.util import find_spec
import os import os
import structlog import structlog
@@ -12,10 +13,13 @@ from python_repositories.interfaces import (
ConnectionAwareInterface, ConnectionAwareInterface,
) )
# Import for mypy type checking only # Handle optional dependencies
if TYPE_CHECKING: if find_spec("redis") is not None:
import redis import redis
from redis.commands.json.path import Path as RedisPath from redis.commands.json.path import Path as RedisPath
else:
redis = None # type: ignore
RedisPath = None # type: ignore
class RedisAdapter( class RedisAdapter(
@@ -25,16 +29,10 @@ class RedisAdapter(
"""Redis adapter exposing basic CRUD functionality.""" """Redis adapter exposing basic CRUD functionality."""
uri_env_var_name: str = "REDIS_URI" uri_env_var_name: str = "REDIS_URI"
path: str = RedisPath.root_path() path: str = "." # JSON root path, updated in __init__
encoding: str = "UTF-8" encoding: str = "UTF-8"
def __init__(self) -> None: 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 # Setup logger
self.logger = structlog.get_logger( self.logger = structlog.get_logger(
self.__class__.__name__, self.__class__.__name__,
@@ -43,6 +41,7 @@ class RedisAdapter(
check_env(self.uri_env_var_name) check_env(self.uri_env_var_name)
# Prepare internal variables # Prepare internal variables
self._client: redis.Redis | None = None self._client: redis.Redis | None = None
self.path: str = RedisPath.root_path()
def __enter__(self) -> RedisAdapter: def __enter__(self) -> RedisAdapter:
"""Enter the context.""" """Enter the context."""