fixed linting errors

TusStorageHandler/app.py

@@ -1,13 +1,15 @@
-from flask import Flask, request, jsonify
-from minio import Minio
-from minio.error import S3Error
+"""Main application module for handling MinIO storage and TUS protocol with Flask."""
+
+import datetime
 import os
 import uuid
-from minio_client import minio_client
+
+from flask import Flask, jsonify, request
 from flask_cors import CORS
-from werkzeug.utils import secure_filename
+from minio.error import S3Error
+from minio_client import minio_client
 from tus_server import TusServer
-import datetime
+from werkzeug.utils import secure_filename
 
 MINIO_BUCKET = os.getenv("MINIO_BUCKET_NAME")
 
@@ -16,102 +18,93 @@
 
 tus_server = TusServer(minio_client, MINIO_BUCKET)
 
-
 @app.route("/", methods=["GET"])
 def home():
-  return jsonify({"message": "Welcome to the MinIO TUS server!"}), 200
-
+    """Endpoint to return a welcome message."""
+    return jsonify({"message": "Welcome to the MinIO TUS server!"}), 200
 
 @app.route("/download", methods=["GET"])
 def download_bucket():
-  try:
-    objects = minio_client.list_objects(MINIO_BUCKET, recursive=True)
-    download_links = []
-    for obj in objects:
-      download_url = minio_client.presigned_get_object(
-        MINIO_BUCKET, obj.object_name, expires=datetime.timedelta(seconds=3600)
-      )
-      download_links.append(download_url)
+    """Endpoint to list all objects in the MinIO bucket.
 
-    # for link in download_links:
-    #     print(f"Download Link: {link}")
+    It uses pre-signed download links.
+    """
+    try:
+        objects = minio_client.list_objects(MINIO_BUCKET, recursive=True)
+        download_links = []
+        for obj in objects:
+            download_url = minio_client.presigned_get_object(
+                MINIO_BUCKET, obj.object_name, expires=datetime.timedelta(seconds=3600)
+            )
+            download_links.append(download_url)
 
-    return "<br>".join(download_links)
-
-  except S3Error as err:
-    print(err)
-    return "Error occurred while fetching download links."
+        return jsonify({"download_links": download_links}), 200
 
+    except S3Error as err:
+        return jsonify({"error": str(err)}), 500
 
 @app.route("/download/<filename>", methods=["GET"])
 def download_file(filename):
-  try:
-    # Generate a pre-signed URL valid for 1 hour
-    download_url = minio_client.presigned_get_object(
-      MINIO_BUCKET,
-      filename,
-      expires=datetime.timedelta(seconds=3600),
-      response_headers={
-        "response-content-disposition": f'attachment; filename="{filename}"'
-      },
-    )
-
-    # Return the pre-signed URL to the client
-    return jsonify({"download_url": download_url})
-
-  except S3Error as e:
-    return jsonify({"error": str(e)}), 500
-
+    """Endpoint to provide a pre-signed URL for downloading a specific file."""
+    try:
+        download_url = minio_client.presigned_get_object(
+            MINIO_BUCKET,
+            filename,
+            expires=datetime.timedelta(seconds=3600),
+            response_headers={
+              "response-content-disposition": f'attachment; filename="{filename}"'
+            },
+        )
+        return jsonify({"download_url": download_url}), 200
+
+    except S3Error as err:
+        return jsonify({"error": str(err)}), 500
 
 @app.route("/list_files", methods=["GET"])
 def list_files():
-  try:
-    objects = minio_client.list_objects(MINIO_BUCKET)
-    files = [obj.object_name for obj in objects]
-    return jsonify({"files": files}), 200
-  except S3Error as e:
-    return jsonify({"error": str(e)}), 500
+    """Endpoint to list all files in the MinIO bucket."""
+    try:
+        objects = minio_client.list_objects(MINIO_BUCKET)
+        files = [obj.object_name for obj in objects]
+        return jsonify({"files": files}), 200
 
+    except S3Error as err:
+        return jsonify({"error": str(err)}), 500
 
 @app.route("/upload", methods=["POST"])
 def upload_file():
-  if "file" not in request.files:
-    return jsonify({"error": "No file part in the request"}), 400
-
-  uploaded_files = request.files.getlist("file")
-  success_responses = []
-  error_responses = []
-
-  for uploaded_file in uploaded_files:
-    if uploaded_file.filename == "":
-      error_responses.append({"error": "Empty filename provided"})
-      continue
-
-    upload_id = str(uuid.uuid4())
-    filename = secure_filename(uploaded_file.filename)
-    file_content = uploaded_file.read()
-
-    if file_content:
-      tus_server.write_file(upload_id, file_content)
-      tus_server.assemble_file(upload_id, filename)
-      location = f"http://{request.host}/upload/{upload_id}"
-      success_responses.append(
-        {
-          "filename": filename,
-          "message": "File uploaded successfully",
-          "Location": location,
-        }
-      )
+    """Endpoint to handle file uploads using the TUS protocol."""
+    if "file" not in request.files:
+        return jsonify({"error": "No file part in the request"}), 400
+
+    uploaded_files = request.files.getlist("file")
+    success_responses = []
+    error_responses = []
+
+    for uploaded_file in uploaded_files:
+        if uploaded_file.filename == "":
+            error_responses.append({"error": "Empty filename provided"})
+            continue
+
+        upload_id = str(uuid.uuid4())
+        filename = secure_filename(uploaded_file.filename)
+        file_content = uploaded_file.read()
+
+        if file_content:
+            tus_server.write_file(upload_id, file_content)
+            tus_server.assemble_file(upload_id, filename)
+            location = f"http://{request.host}/upload/{upload_id}"
+            success_responses.append({"filename": filename, "location": location})
+        else:
+            error_responses.append({
+                "filename": filename,
+                "error": "No file content provided"}
+            )
+
+    if error_responses:
+        return jsonify({"errors": error_responses}), 400
     else:
-      error_responses.append(
-        {"filename": filename, "error": "No file content provided"}
-      )
-
-  if error_responses:
-    return jsonify({"errors": error_responses}), 400
-  else:
-    return jsonify({"success": success_responses}), 201
-
+        return jsonify({"success": success_responses}), 201
 
 if __name__ == "__main__":
-  app.run(debug=False)
+    app.run(debug=False)

TusStorageHandler/minio_client.py

@@ -1,3 +1,22 @@
+"""Minio Client Configuration Module.
+
+It initializes a Minio client using environment variables
+loaded via dotenv. It also provides a function to create a new Minio
+bucket if it does not already exist.
+
+Usage:
+    - Ensure environment variables are set in a .env file:
+      - MINIO_HOSTNAME
+      - MINIO_PORT
+      - MINIO_ACCESS_KEY
+      - MINIO_SECRET_KEY
+      - MINIO_IS_SECURE (optional, defaults to 'true')
+      - MINIO_BUCKET_NAME
+
+    - Import `minio_client` to access:
+      - `minio_client`: Initialized Minio client instance
+      - `create_bucket()`: Function to create a new bucket if not exists.
+"""
 import os
 
 from dotenv import load_dotenv
@@ -22,6 +41,11 @@
 
 
 def create_bucket():
+  """Creates a new bucket if it does not already exist.
+
+  This function checks if the bucket specified by MINIO_BUCKET_NAME exists. 
+  If it does not exist, it creates a new bucket using the Minio client instance.
+  """
   if not minio_client.bucket_exists(MINIO_BUCKET_NAME):
     minio_client.make_bucket(MINIO_BUCKET_NAME)
 

TusStorageHandler/tus_server.py

@@ -1,33 +1,79 @@
+"""TusServer Module defines the TusServer class for handling files."""
+
 import os
+
 from minio.error import S3Error
 
 
 class TusServer:
+  """Initializes a new instance of the TusServer class.
+
+  Args:
+      minio_client (Minio): The Minio client object used for file operations.
+      bucket_name (str): The name of the Minio bucket to operate on.
+  """
   def __init__(self, minio_client, bucket_name):
+    """Initializes a new instance of the TusServer class.
+
+    Args:
+        minio_client (Minio): The Minio client object used for file operations.
+        bucket_name (str): The name of the Minio bucket to operate on.
+
+    """
     self.minio_client = minio_client
     self.bucket_name = bucket_name
 
   def get_file_path(self, upload_id):
+    """Returns the file path for the given upload_id within the 'uploads' directory.
+
+    Args:
+        upload_id (str): Unique identifier for the upload.
+
+    Returns:
+        str: File path within the 'uploads' directory corresponding to the upload_id.
+
+    """
     return os.path.join("uploads", upload_id)
 
   def write_file(self, upload_id, file_content):
+    """Writes content to a file located at the specified upload_id location.
+
+    Args:
+        upload_id (str): Unique identifier for the upload location.
+        file_content (bytes): Content to be written to the file.
+
+    Raises:
+        IOError: If there is an error writing to the file.
+
+    """
     file_path = self.get_file_path(upload_id)
     if not os.path.exists(file_path):
       os.makedirs(os.path.dirname(file_path), exist_ok=True)
     with open(file_path, "wb") as f:
       f.write(file_content)
 
   def assemble_file(self, upload_id, filename):
+    """Uploads a file to Minio storage after assembling it from a temporary location.
+
+    Args:
+        upload_id (str): Unique identifier for the upload.
+        filename (str): Name to assign to the uploaded file in Minio.
+
+    Raises:
+        Exception: If the file cannot be uploaded to Minio or if the file specified 
+                   by `upload_id` does not exist.
+
+    """
     file_path = self.get_file_path(upload_id)
     if os.path.exists(file_path):
-      try:
-        # Upload the file to Minio
-        self.minio_client.fput_object(self.bucket_name, filename, file_path)
-      except S3Error as err:
-        print(err)
-        raise Exception("Failed to upload file")
-      finally:
-        # Remove the temporary file
-        os.remove(file_path)
+        try:
+            # Upload the file to Minio
+            self.minio_client.fput_object(self.bucket_name, filename, file_path)
+        except S3Error as err:
+            print(f"Failed to upload file to Minio: {err}")
+            raise Exception("Failed to upload file to Minio") from err
+        finally:
+            # Remove the temporary file
+            os.remove(file_path)
     else:
-      raise Exception(f"File not found for upload_id: {upload_id}")
+        raise Exception(f"File not found for upload_id: {upload_id}")