def get_text_embeddings()

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

• 25 lines of code
• 6 McCabe index (conditional complexity)

def get_text_embeddings(text):
    """
    Fetches text embeddings from a text embedding API.

    Args:
        text: The input text.

    Returns:
        The text embeddings as a JSON object.

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

        response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)

        return response.json()["text_embeds"]

    except requests.exceptions.HTTPError as e:
        logger.exception("Error fetching text embedding: %s", e)
        raise

    except requests.exceptions.RequestException as e:
        logger.exception("Error fetching text embedding: %s", e)
        raise requests.exceptions.HTTPError(
            "Error fetching text embedding", response=requests.Response()
        ) from e

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