def download_image_from_gcs()

in use-cases/rag-pipeline/embedding-models/multimodal-embedding/src/blip2_server.py [0:0]

• 15 lines of code
• 5 McCabe index (conditional complexity)

def download_image_from_gcs(gcs_uri):
    """Downloads an image file from Google Cloud Storage (GCS).

    Args:
        gcs_uri: The GCS URI of the image file (e.g., 'gs://bucket-name/path/to/image.jpg').

    Returns:
        A PIL.Image object.

    Raises:
        ValueError: If the GCS URI is invalid or if there is an error downloading the image.
    """
    try:
        # Validate the GCS URI
        if not gcs_uri.startswith("gs://"):
            raise ValueError("Invalid GCS URI")

        # Extract bucket name and object name from the URI
        bucket_name, object_name = gcs_uri[5:].split("/", 1)

        # Initialize a GCS client
        storage_client = storage.Client()

        # Get the bucket and blob (object)
        bucket = storage_client.bucket(bucket_name)
        blob = bucket.blob(object_name)

        # Download the image into a BytesIO object
        imgf = io.BytesIO()
        blob.download_to_file(imgf)
        imgf.seek(0)

        # Open the image using PIL
        img = Image.open(imgf).convert("RGB")
        return img

    except Exception as e:
        raise ValueError(f"Error downloading image from GCS: {e}")