Add Pylint configuration and implement Azure Function for WSI to DZI processing

- Create GitHub Actions workflow for linting Python code with Pylint.
- Add Pylint configuration file with max line length set to 150.
- Update README to reflect project name and clarify input formats.
- Implement functions for downloading blobs, converting images to DZI, and cleaning up temporary files.
- Integrate AzCopy for uploading DZI outputs to Azure Blob Storage.
- Enhance logging for better traceability during processing.

Author: gsaini

.github/workflows/pylint.yml

@@ -0,0 +1,25 @@
+name: Lint Python with Pylint
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+          pip install pylint
+      - name: Run Pylint
+        run: |
+          pylint $(git ls-files '*.py')

.pylintrc

@@ -0,0 +1,2 @@
+[FORMAT]
+max-line-length=150

README.md

@@ -1,4 +1,4 @@
-# wsi-slides-dzi-processor Azure Function
+# WSI Slide Image to DZI Processor
 
 ![Python](https://img.shields.io/badge/python-3670A0?style=for-the-badge&logo=python&logoColor=ffdd54)
 ![Azure](https://img.shields.io/badge/azure-%230072C6.svg?style=for-the-badge&logo=microsoftazure&logoColor=white)
@@ -10,9 +10,9 @@
 This Azure Function automatically processes Whole Slide Images (WSI) uploaded to Azure Blob Storage, converting them into Deep Zoom Image (DZI) format using `pyvips`. It is designed to handle very large files (up to several GB) efficiently and is suitable for digital pathology, microscopy, and similar domains.
 
 - **Trigger:** Event Grid (on blob upload)
-- **Input:** Image blob (e.g., `.jpg`, `.tif`) in a monitored container
+- **Input:** Image blob (e.g., `.svs`, `.tiff`) in a monitored container
 - **Output:** DZI tiles and metadata uploaded to a designated output container
-- **Tech Stack:** Python, Azure Functions, Azure Blob Storage, pyvips
+- **Tech Stack:** Python, Azure Functions, Azure Blob Storage, AzCopy, pyvips
 
 > **Note:**
 > This function is built and deployed using Docker because the `libvips` library (required by `pyvips`) is not supported on standard Azure Functions hosting options such as Flex Consumption, Consumption, or App Service plans. Docker allows you to include all necessary native dependencies for reliable execution on the Premium plan or in a custom container environment.
@@ -28,6 +28,12 @@ This Azure Function automatically processes Whole Slide Images (WSI) uploaded to
 ```
 ├── Dockerfile
 ├── function_app.py
+├── parse_event_blob_url.py
+├── parse_container_and_blob.py
+├── download_blob_to_temp.py
+├── convert_to_dzi.py
+├── upload_with_azcopy.py
+├── cleanup_temp_files.py
 ├── host.json
 ├── local.settings.json
 ├── requirements.txt
@@ -155,7 +161,13 @@ Create a file named `event.json` with the following content:
     "subject": "/blobServices/default/containers/your-container/blobs/your-image.jpg",
     "eventTime": "2025-06-27T00:00:00.000Z",
     "data": {
-      "url": "https://<your-storage-account>.blob.core.windows.net/your-container/your-image.jpg"
+      "url": "https://<your-storage-account>.blob.core.windows.net/your-container/your-image.jpg",
+      "api": "PutBlob",
+      "contentType": "image/jpeg",
+      "contentLength": 12345678,
+      "blobType": "BlockBlob",
+      "blobTier": "Hot",
+      "metadata": {}
     },
     "dataVersion": "",
     "metadataVersion": "1"
@@ -181,4 +193,37 @@ curl -X POST "http://localhost:7071/runtime/webhooks/EventGrid?functionName=blob
 - Make sure all dependencies are installed and the correct Python version is used.
 
 ---
-*For more, see the Azure Functions [local development docs](https://learn.microsoft.com/azure/azure-functions/functions-develop-local).*
+
+## AzCopy Authentication: Managed Identity and SAS Token
+
+This function uploads DZI output directories to Azure Blob Storage using AzCopy. Two authentication methods are supported:
+
+### 1. Managed Identity (Recommended for Production)
+- **How it works:**
+  - The Azure Function runs with a User-Assigned or System-Assigned Managed Identity.
+  - AzCopy uses the identity to obtain an OAuth token and authenticate to Azure Blob Storage.
+- **Requirements:**
+  - The Function App's Managed Identity must have at least `Storage Blob Data Contributor` role on the target storage account or container.
+  - No secrets or connection strings are required in code or environment variables.
+- **How to use:**
+  - Ensure the Function App is assigned a Managed Identity in Azure.
+  - Grant the identity access to the storage account/container.
+  - **Important:** Set the environment variable `AZCOPY_AUTO_LOGIN_TYPE` to `MSI`.
+  - AzCopy will automatically use the identity for authentication when running inside Azure.
+
+### 2. SAS Token (For Local Development or Special Cases)
+- **How it works:**
+  - AzCopy authenticates using a Shared Access Signature (SAS) token appended to the destination Blob Storage URL.
+- **Requirements:**
+  - A valid SAS token with write permissions for the target container or directory.
+- **How to use:**
+  - Generate a SAS token for the storage account or container.
+  - Append the SAS token to the destination URL in the AzCopy command (e.g., `https://<account>.blob.core.windows.net/<container>?<sas-token>`).
+  - This method is useful for local testing or scenarios where Managed Identity is not available.
+
+> **Best Practice:**
+> Use Managed Identity for all production deployments to avoid secret management and improve security. SAS tokens should only be used for local development or temporary access.
+
+Please feel free to reach out to me if you have any questions or need further assistance with the WSI Slide Image to DZI Processor project. Your feedback and contributions are always welcome!
+
+---

cleanup_temp_files.py

@@ -0,0 +1,24 @@
+"""
+cleanup_temp_files.py: Removes temporary files and directories.
+"""
+import os
+import shutil
+import logging
+
+def cleanup_temp_files(*paths):
+    """
+    Removes temporary files and directories.
+    Args:
+        *paths: Paths to files or directories to remove.
+    """
+    logger = logging.getLogger("helpers.cleanup_temp_files")
+    for path in paths:
+        try:
+            if os.path.isdir(path):
+                shutil.rmtree(path)
+                logger.info("Cleaned up temp directory: %s", path)
+            elif os.path.isfile(path):
+                os.remove(path)
+                logger.info("Cleaned up temp file: %s", path)
+        except (OSError, shutil.Error) as exc:
+            logger.warning("Failed to clean up %s: %s", path, exc)

convert_to_dzi.py

@@ -0,0 +1,30 @@
+"""
+convert_to_dzi.py: Converts an image file to Deep Zoom Image (DZI) format using pyvips.
+"""
+import tempfile
+import os
+import logging
+from typing import Optional
+import pyvips
+
+def convert_to_dzi(input_path: str, blob_name: str) -> Optional[str]:
+    """
+    Converts an image file to Deep Zoom Image (DZI) format using pyvips.
+    Args:
+        input_path: Path to the input image file.
+        blob_name: The original blob name (used for output naming).
+    Returns:
+        The path to the DZI output directory, or None if conversion fails.
+    """
+    logger = logging.getLogger("helpers.convert_to_dzi")
+    try:
+        dzi_output_dir = tempfile.mkdtemp()
+        dzi_basename = os.path.splitext(os.path.basename(blob_name))[0]
+        dzi_output_path = os.path.join(dzi_output_dir, dzi_basename)
+        logger.info("Converting to DZI: %s", dzi_output_path)
+        image = pyvips.Image.new_from_file(input_path, access='sequential')
+        image.dzsave(dzi_output_path, tile_size=512)
+        return dzi_output_dir
+    except (pyvips.Error, OSError) as exc:
+        logger.error("DZI conversion failed: %s", exc)
+        return None

download_blob_to_temp.py

@@ -0,0 +1,48 @@
+"""
+download_blob_to_temp.py: Downloads a blob to a temporary file using streaming.
+"""
+import tempfile
+import os
+import logging
+from typing import Optional
+from azure.storage.blob import BlobServiceClient
+from azure.core.exceptions import AzureError
+
+def download_blob_to_temp(blob_service_client: BlobServiceClient, container_name: str, blob_name: str) -> Optional[str]:
+    """
+    Downloads a blob to a temporary file using streaming (suitable for large files).
+    Args:
+        blob_service_client: The BlobServiceClient instance.
+        container_name: The name of the container.
+        blob_name: The name of the blob.
+    Returns:
+        The path to the temporary file, or None if download fails.
+    """
+    logger = logging.getLogger("helpers.download_blob_to_temp")
+    try:
+        container_client = blob_service_client.get_container_client(container_name)
+        exists = container_client.exists()
+        if not exists:
+            logger.error(
+                "Container '%s' does not exist in the storage account.", container_name
+            )
+            return None
+        blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)
+        with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(blob_name)[1]) as temp_blob:
+            logger.info("Streaming blob to temp file: %s", temp_blob.name)
+            download_stream = blob_client.download_blob()
+            chunk_size = 8 * 1024 * 1024  # 8 MB
+            offset = 0
+            while True:
+                chunk = download_stream.read(chunk_size)
+                if not chunk:
+                    break
+                temp_blob.write(chunk)
+                offset += len(chunk)
+            logger.info(
+                "Completed streaming blob to: %s, total bytes: %d", temp_blob.name, offset
+            )
+            return temp_blob.name
+    except (OSError, AzureError) as exc:
+        logger.error("Failed to download blob (streaming): %s", exc)
+        return None

function_app.py

@@ -1,120 +1,72 @@
-import azure.functions as func
+"""
+function_app.py: Azure Function entrypoint for WSI to DZI processing.
+"""
+
+import asyncio
 import logging
 import os
-import asyncio
+
+import azure.functions as func
 from azure.storage.blob import BlobServiceClient
-import pyvips
-import subprocess
+
+from cleanup_temp_files import cleanup_temp_files
+from convert_to_dzi import convert_to_dzi
+from download_blob_to_temp import download_blob_to_temp
+from parse_container_and_blob import parse_container_and_blob
+from parse_event_blob_url import parse_event_blob_url
+from upload_with_azcopy import upload_with_azcopy
 
 app = func.FunctionApp()
 
+
 @app.event_grid_trigger(arg_name="event")
 async def blob_to_dzi_eventgrid_trigger(event: func.EventGridEvent):
+    """
+    Azure Function entrypoint for processing blob events:
+    - Parses the event for the blob URL
+    - Downloads the blob to a temp file
+    - Converts the blob to DZI format
+    - Uploads the DZI output to Azure Blob Storage using AzCopy
+    - Cleans up temp files and directories
+    """
     logger = logging.getLogger("blob_to_dzi_eventgrid_trigger")
-    logger.info(f"Received Event: id={getattr(event, 'id', None)}, subject={getattr(event, 'subject', None)}")
-    try:
-        event_data = event.get_json()
-        blob_url = event_data.get('url')
-        logger.info(f"Parsed blob URL: {blob_url}")
-    except Exception as e:
-        logger.error(f"Error accessing event.get_json(): {e}")
-        blob_url = None
+    logger.info(
+        "Received Event: id=%s, subject=%s",
+        getattr(event, "id", None),
+        getattr(event, "subject", None),
+    )
+
+    blob_url = parse_event_blob_url(event)
     if not blob_url:
-        logger.error('No blob URL found in event data.')
+        logger.error("No blob URL found in event data.")
         return
 
-    # Parse storage account, container, and blob name from URL
-    from urllib.parse import urlparse
-    parsed = urlparse(blob_url)
-    path_parts = parsed.path.lstrip('/').split('/', 1)
-    if len(path_parts) != 2:
-        logger.error(f'Could not parse container and blob name from URL: {blob_url}')
+    parsed = parse_container_and_blob(blob_url)
+    if not parsed:
         return
-    container_name, blob_name = path_parts
-    logger.info(f"Container: {container_name}, Blob: {blob_name}")
+    container_name, blob_name = parsed
+    logger.info("Container: %s, Blob: %s", container_name, blob_name)
 
-    # Get connection string from environment
-    conn_str = os.environ.get('AzureWebJobsStorage')
+    conn_str = os.environ.get("AzureWebJobsStorage")
     if not conn_str:
-        logger.error('AzureWebJobsStorage not set in environment.')
+        logger.error("AzureWebJobsStorage not set in environment.")
         return
 
     blob_service_client = BlobServiceClient.from_connection_string(conn_str)
-    blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)
-
-    # Download blob to temp file (streaming, suitable for very large files)
-    import tempfile
-    logger.info(f"Attempting to download from container: '{container_name}', blob: '{blob_name}' (streaming mode)")
-    try:
-        # Check if container exists
-        container_client = blob_service_client.get_container_client(container_name)
-        exists = await asyncio.to_thread(container_client.exists)
-        if not exists:
-            logger.error(f"Container '{container_name}' does not exist in the storage account.")
-            return
-        with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(blob_name)[1]) as temp_blob:
-            logger.info(f"Streaming blob to temp file: {temp_blob.name}")
-            download_stream = await asyncio.to_thread(blob_client.download_blob)
-            # Stream the blob in chunks to avoid high memory usage
-            chunk_size = 8 * 1024 * 1024  # 8 MB
-            offset = 0
-            while True:
-                chunk = await asyncio.to_thread(download_stream.read, chunk_size)
-                if not chunk:
-                    break
-                temp_blob.write(chunk)
-                offset += len(chunk)
-            temp_blob_path = temp_blob.name
-            logger.info(f"Completed streaming blob to: {temp_blob_path}, total bytes: {offset}")
-    except Exception as e:
-        logger.error(f"Failed to download blob (streaming): {e}")
+    temp_blob_path = await asyncio.to_thread(
+        download_blob_to_temp, blob_service_client, container_name, blob_name
+    )
+    if not temp_blob_path:
         return
 
-    # Convert to DZI using pyvips
-    try:
-        dzi_output_dir = tempfile.mkdtemp()
-        dzi_basename = os.path.splitext(os.path.basename(blob_name))[0]
-        dzi_output_path = os.path.join(dzi_output_dir, dzi_basename)
-        logger.info(f"Converting to DZI: {dzi_output_path}")
-        image = await asyncio.to_thread(pyvips.Image.new_from_file, temp_blob_path, access='sequential')
-        await asyncio.to_thread(
-            image.dzsave,
-            dzi_output_path,
-            tile_size=512  # Recommended for OpenSeadragon, fewer files, good performance
-        )
-    except Exception as e:
-        logger.error(f"DZI conversion failed: {e}")
+    dzi_output_dir = await asyncio.to_thread(convert_to_dzi, temp_blob_path, blob_name)
+    if not dzi_output_dir:
         return
 
-    # Upload DZI files and all subdirectory files to the destination container using AzCopy
-    def upload_with_azcopy(local_dir):
-        dest_url = os.environ.get('DZI_UPLOAD_DEST_URL')
-        cmd = [
-            "azcopy", "copy",
-             f"{local_dir}/*",
-            dest_url,
-            "--recursive=true"
-        ]
-        logger.info(f"AzCopy command: azcopy copy {cmd[2]} {dest_url} --recursive=true (using SAS token)")
-        result = subprocess.run(cmd, capture_output=True, text=True)
-        logger.info(f"AzCopy stdout: {result.stdout}")
-        logger.info(f"AzCopy stderr: {result.stderr}")
-        logger.info(f"AzCopy returncode: {result.returncode}")
-        if result.returncode == 0:
-            logger.info(f"AzCopy upload successful for {cmd[2]}")
-        else:
-            logger.error(f"AzCopy failed with exit code {result.returncode} for {cmd[2]}")
-
-    upload_with_azcopy(dzi_output_dir)
-
-    # Clean up temp files and directories
-    import shutil
-    try:
-        if os.path.exists(temp_blob_path):
-            os.remove(temp_blob_path)
-            logger.info(f"Cleaned up temp file: {temp_blob_path}")
-        if os.path.exists(dzi_output_dir):
-            shutil.rmtree(dzi_output_dir)
-            logger.info(f"Cleaned up temp directory: {dzi_output_dir}")
-    except Exception as e:
-        logger.warning(f"Failed to clean up temp files or directories: {e}")
+    dest_url = os.environ.get("DZI_UPLOAD_DEST_URL")
+    if not dest_url:
+        logger.error("DZI_UPLOAD_DEST_URL environment variable not set!")
+        cleanup_temp_files(temp_blob_path, dzi_output_dir)
+        return
+    upload_with_azcopy(dzi_output_dir, dest_url, logger)
+    cleanup_temp_files(temp_blob_path, dzi_output_dir)