def database_query(
        self, query: str, params: Optional[dict] = None
    ) -> List[Dict[str, Any]]:
        """
        This method sends a query to the Neptune Analytics graph
        and returns the results as a list of dictionaries.

        Args:
            query (str): The openCypher query to execute.
            params (dict, optional): Dictionary of query parameters. Defaults to {}.

        Returns:
            List[Dict[str, Any]]: List of dictionaries containing the query results.

        """
        try:
            logger.debug(f"query() query: {query} parameters: {json.dumps(params)}")
            resp = self._client.execute_query(
                graphIdentifier=self.graph_identifier,
                queryString=query,
                parameters=params,
                language="OPEN_CYPHER",
            )
            return json.loads(resp["payload"].read().decode("UTF-8"))["results"]
        except Exception as e:
            raise NeptuneVectorQueryException(
                {
                    "message": "An error occurred while executing the query.",
                    "details": str(e),
                }
            )

        This method sends a query to the Neptune Analytics graph
        and returns the results as a list of dictionaries.

        Args:
            query (str): The openCypher query to execute.
            params (dict, optional): Dictionary of query parameters. Defaults to {}.

        Returns:
            List[Dict[str, Any]]: List of dictionaries containing the query results.

        
llama-index-integrations/vector_stores/llama-index-vector-stores-neptune/llama_index/vector_stores/neptune/base.py