def get_image_embeddings()

in use-cases/rag-pipeline/backend/src/generate_embeddings.py [0:0]

• 26 lines of code
• 7 McCabe index (conditional complexity)

def get_image_embeddings(image_uri):
    """
    Fetches image embeddings from an image embedding API.

    Args:
        image_uri: The URI of the image.

    Returns:
        The image embeddings as a JSON object.

    Raises:
        requests.exceptions.HTTPError: If there is an error fetching the image embeddings
                                       or the API returns an invalid response.
    """
    try:
        response = requests.post(
            IMAGE_API_ENDPOINT,
            json={"image_uri": image_uri},
            headers={"Content-Type": "application/json"},
            timeout=100,
        )

        # This will raise an HTTPError for bad responses (4xx or 5xx)
        response.raise_for_status()

        image_embeddings = response.json()["image_embeds"]
        return image_embeddings

    except requests.exceptions.HTTPError as e:
        # Reraise HTTPError for better error handling
        logger.exception("Error fetching image embedding: %s", e)
        raise

    except requests.exceptions.RequestException as e:
        # For other request errors, re-raise as an HTTPError
        logger.exception("Invalid response from image embedding API: %s", e)
        raise requests.exceptions.HTTPError(
            "Error fetching image embedding", response=requests.Response()
        ) from e

    except (ValueError, TypeError) as e:
        # Handle potential JSON decoding errors
        logger.exception(
            "Not able to decode received json from image embedding API: %s", e
        )
        raise requests.exceptions.HTTPError(
            "Invalid response from image embedding API", response=requests.Response()
        ) from e