proposal

cloud_storage_handler/api/elixircloud/csh/client.py

@@ -0,0 +1,61 @@
+"""MinIO client class and convenience functions."""
+
+import logging
+from typing import Type
+
+from connexion import FlaskApp
+from minio import Minio
+
+from cloud_storage_handler.api.elixircloud.csh.models import MinioConfig
+
+logger = logging.getLogger(__name__)
+
+
+class MinioClient:
+    """Client for MinIO operations.
+
+    Wraps ``minio.Minio`` and adds convenience methods.
+
+    Attributes:
+        config: MinIO configuration.
+        client: MinIO client instance.
+    """
+
+    def __init__(self, config: MinioConfig) -> None:
+        """Class constructor.
+
+        Args:
+            config: MinIO configuration.
+        """
+        self.config: MinioConfig = config
+        self.client: Type[Minio] = Minio(
+            endpoint=f"{config.hostname}:{config.port}",
+            access_key=config.access_key,
+            secret_key=config.secret_key,
+            secure=config.secure,
+        )
+
+    def create_bucket(self) -> None:
+        """Create bucket if it does not exist."""
+        if not self.client.bucket_exists(self.config.bucket_name):
+            self.client.make_bucket(self.config.bucket_name)
+            logger.info(f"Bucket '{self.config.bucket_name}' created.")
+        else:
+            logger.debug(f"Bucket '{self.config.bucket_name}' already exists.")
+
+
+def register_minio_client(app: FlaskApp) -> FlaskApp:
+    """Register MinIO client and create bucket.
+
+    Args:
+        app: FOCA app instance.
+
+    Returns:
+        FOCA app instance with MinIO client instance added to config.
+    """
+    minio_config = app.app.config.foca.custom.minio
+    minio_client = MinioClient(config=minio_config)
+    minio_client.create_bucket()
+    minio_config.client = minio_client
+    logger.info("MinIO client registered.")
+    return app

cloud_storage_handler/api/elixircloud/csh/models.py

@@ -1,21 +1,24 @@
 """Model for MinIO Configuration."""
 
-from typing import Annotated
+from typing import Annotated, Optional, Type
 
+from minio import Minio
 from pydantic import BaseModel, conint, constr
 
 
 class MinioConfig(BaseModel):
     """Configuration for MinIO.
 
-    Attributes:  
+    Attributes:
         hostname: The name of the host where the MinIO server is running.
         port: The port on which the MinIO server is running. Must be between 1
             and 65535.
         access_key: The access key used for authentication with MinIO.
         secret_key: The secret key used for authentication with MinIO.
         bucket_name: The name of the bucket where files are stored. Must be at
             least 1 character long.
+        secure: Whether to use TLS connection to storage service or not.
+        client: Client instance.
 
     Examples:
         MinioConfig(
@@ -24,7 +27,7 @@ class MinioConfig(BaseModel):
             access_key="minioadmin",
             secret_key="minioadmin",
             bucket_name="files",
-            secure=False
+            secure=False,
         )
     """
 
@@ -34,3 +37,4 @@ class MinioConfig(BaseModel):
     secret_key: str = "minioadmin"
     bucket_name: Annotated[str, constr(min_length=1)] = "files"
     secure: bool = False
+    client: Optional[Type[Minio]] = None

cloud_storage_handler/main.py

@@ -1,87 +1,44 @@
 """API server entry point."""
 
-import logging
 import os
-import sys
 from pathlib import Path
 
 from connexion import FlaskApp
 from foca import Foca
-from minio import Minio
-from pydantic import ValidationError
 
-from cloud_storage_handler.api.elixircloud.csh.models import MinioConfig
-from cloud_storage_handler.custom_config import CustomConfig
+from cloud_storage_handler.api.elixircloud.csh.client import (
+    register_minio_client,
+)
 
-logger = logging.getLogger(__name__)
 
+def init_app() -> FlaskApp:
+    """Initialize and return the FOCA app.
 
-def init_app() -> tuple[FlaskApp, MinioConfig]:
-    """Initialize and return the FOCA app and MinIO configuration."""
-    # Determine the configuration path
-    if config_path_env := os.getenv("CSH_FOCA_CONFIG_PATH"):
-        config_path = Path(config_path_env).resolve()
-    else:
-        config_path = (
-            Path(__file__).parents[1] / "deployment" / "config.yaml"
-        ).resolve()
+    This function initializes the FOCA app by loading the configuration
+    from the environment variable `CSH_FOCA_CONFIG_PATH` if set, or from
+    the default path if not.
 
-    # Check if the configuration file exists
-    if not config_path.exists():
-        raise FileNotFoundError(f"Config file not found at: {config_path}")
+    Returns:
+        A Connexion application instance.
+    """
+    config_path = Path(
+        os.getenv(
+            "CSH_FOCA_CONFIG_PATH",
+            Path(__file__).parents[1] / "deployment" / "config.yaml",
+        )
+    ).resolve()
 
     foca = Foca(
         config_file=config_path,
         custom_config_model="cloud_storage_handler.custom_config.CustomConfig",
     )
-    foca_config = foca.conf
-    try:
-        custom_config_data = foca_config.custom.dict()
-        custom_config = CustomConfig(**custom_config_data)
-    except ValidationError as e:
-        logger.error(f"Error parsing custom configuration: {e}")
-        raise
-    minio_config = custom_config.minio
-
-    # Create the Connexion FlaskApp instance
-    return foca.create_app(), minio_config
+    return foca.create_app()
 
 
 def main() -> None:
     """Run FOCA application."""
-    try:
-        app, minio_config = init_app()
-    except Exception as e:
-        logger.error(f"Unexpected error during initialization: {e}")
-        sys.exit(1)
-    foca_app = app.app
-
-    # Initialize MinIO client
-    try:
-        minio_client = Minio(
-            endpoint=f"{minio_config.hostname}:{minio_config.port}",
-            access_key=minio_config.access_key,
-            secret_key=minio_config.secret_key,
-            secure=minio_config.secure,
-        )
-        foca_app.config["minio_client"] = minio_client
-    except Exception as e:
-        logger.error(f"Error initializing MinIO client: {e}")
-        return
-
-    # Check if the bucket exists and create it if it doesn't
-    bucket_name = minio_config.bucket_name
-    try:
-        if not minio_client.bucket_exists(bucket_name):
-            minio_client.make_bucket(bucket_name)
-            logger.info(f"Bucket '{bucket_name}' created.")
-        else:
-            logger.info(f"Bucket '{bucket_name}' already exists.")
-    except Exception as e:
-        logger.error(f"Error checking or creating bucket: {e}")
-        return
-
-    # Start the Flask app
+    app = init_app()
+    app = register_minio_client(app)
     app.run(port=app.port)