def vertex_ai_search()

in search/cloud-function/python/main.py [0:0]
31 lines of code
14 McCabe index (conditional complexity)

def vertex_ai_search(http_request: Request) -> tuple[Any, int, dict[str, str]]:
    """
    Handle HTTP requests for Vertex AI Search.

    This function processes incoming HTTP requests, performs the search using
    the VertexAISearchClient, and returns the results. It handles CORS, validates
    the request, and manages potential errors.

    Args:
        http_request (flask.Request): The request object.
        <https://flask.palletsprojects.com/en/1.1.x/api/#incoming-request-data>

    Returns:
        Tuple[Any, int, Dict[str, str]]: A tuple containing the response body,
        status code, and headers. This output will be turned into a Response
        object using `make_response`
        <https://flask.palletsprojects.com/en/1.1.x/api/#flask.make_response>.
    """
    # Set CORS headers for the preflight request
    if http_request.method == "OPTIONS":
        # Allows GET requests from any origin with the Content-Type
        # header and caches preflight response for an 3600s
        headers = {
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Methods": "GET",
            "Access-Control-Allow-Headers": "Content-Type",
            "Access-Control-Max-Age": "3600",
        }
        return ("", 204, headers)

    # Set CORS headers for all responses
    headers = {"Access-Control-Allow-Origin": "*"}

    def create_error_response(
        message: str, status_code: int
    ) -> tuple[Any, int, dict[str, str]]:
        """Standardize the error responses with common headers."""
        return (jsonify({"error": message}), status_code, headers)

    # Handle the request and get the search_term
    request_json = http_request.get_json(silent=True)
    request_args = http_request.args

    if request_json and "search_term" in request_json:
        search_term = request_json["search_term"]
    elif request_args and "search_term" in request_args:
        search_term = request_args["search_term"]
    else:
        return create_error_response("No search term provided", 400)

    # Handle the Vertex AI Search and return JSON
    try:
        results = vertex_ai_search_client.search(search_term)
        return (jsonify(results), 200, headers)
    except GoogleAPICallError as e:
        return create_error_response(
            f"Error calling Vertex AI Search API: {str(e)}", 500
        )
    except ValueError as e:
        return create_error_response(f"Invalid input: {str(e)}", 400)