Run a query on a Cassandra cluster and return a dictionary.

    This function should not be used asynchronously for SELECTs -- it will not
    return anything and we don't currently have a mechanism for handling a future
    that will return results.

    :param query:          The query to execute.
    :type  query:          str
    :param statement_name: Name to assign the prepared statement in the __context__ dictionary
    :type  statement_name: str
    :param statement_arguments: Bind parameters for the SQL statement
    :type  statement_arguments: list[str]
    :param async:           Run this query in asynchronous mode
    :type  async:           bool
    :param callback_errors: Function to call after query runs if there is an error
    :type  callback_errors: Function callable
    :param contact_points: The Cassandra cluster addresses, can either be a string or a list of IPs.
    :type  contact_points: str | list[str]
    :param cql_user:       The Cassandra user if authentication is turned on.
    :type  cql_user:       str
    :param cql_pass:       The Cassandra user password if authentication is turned on.
    :type  cql_pass:       str
    :param port:           The Cassandra cluster port, defaults to None.
    :type  port:           int
    :param params:         The parameters for the query, optional.
    :type  params:         str
    :return:               A dictionary from the return values of the query
    :rtype:                list[dict]


    CLI Example:

    .. code-block:: bash

        # Insert data asynchronously
        salt this-node cassandra_cql.cql_query_with_prepare "name_insert" "INSERT INTO USERS (first_name, last_name) VALUES (?, ?)" \
            statement_arguments=['John','Doe'], asynchronous=True

        # Select data, should not be asynchronous because there is not currently a facility to return data from a future
        salt this-node cassandra_cql.cql_query_with_prepare "name_select" "SELECT * FROM USERS WHERE first_name=?" \
            statement_arguments=['John']

def cql_query_with_prepare(query, statement_name, statement_arguments, callback_errors=None, contact_points=None,
                           port=None, cql_user=None, cql_pass=None, **kwargs):
    '''
    Run a query on a Cassandra cluster and return a dictionary.

    This function should not be used asynchronously for SELECTs -- it will not
    return anything and we don't currently have a mechanism for handling a future
    that will return results.

    :param query:          The query to execute.
    :type  query:          str
    :param statement_name: Name to assign the prepared statement in the __context__ dictionary
    :type  statement_name: str
    :param statement_arguments: Bind parameters for the SQL statement
    :type  statement_arguments: list[str]
    :param async:           Run this query in asynchronous mode
    :type  async:           bool
    :param callback_errors: Function to call after query runs if there is an error
    :type  callback_errors: Function callable
    :param contact_points: The Cassandra cluster addresses, can either be a string or a list of IPs.
    :type  contact_points: str | list[str]
    :param cql_user:       The Cassandra user if authentication is turned on.
    :type  cql_user:       str
    :param cql_pass:       The Cassandra user password if authentication is turned on.
    :type  cql_pass:       str
    :param port:           The Cassandra cluster port, defaults to None.
    :type  port:           int
    :param params:         The parameters for the query, optional.
    :type  params:         str
    :return:               A dictionary from the return values of the query
    :rtype:                list[dict]


    CLI Example:

    .. code-block:: bash

        # Insert data asynchronously
        salt this-node cassandra_cql.cql_query_with_prepare "name_insert" "INSERT INTO USERS (first_name, last_name) VALUES (?, ?)" \
            statement_arguments=['John','Doe'], asynchronous=True

        # Select data, should not be asynchronous because there is not currently a facility to return data from a future
        salt this-node cassandra_cql.cql_query_with_prepare "name_select" "SELECT * FROM USERS WHERE first_name=?" \
            statement_arguments=['John']
    '''
    # Backward-compatibility with Python 3.7: "async" is a reserved word
    asynchronous = kwargs.get('async', False)
    try:
        cluster, session = _connect(contact_points=contact_points, port=port,
                                    cql_user=cql_user, cql_pass=cql_pass)
    except CommandExecutionError:
        log.critical('Could not get Cassandra cluster session.')
        raise
    except BaseException as e:
        log.critical('Unexpected error while getting Cassandra cluster session: %s', e)
        raise

    if statement_name not in __context__['cassandra_cql_prepared']:
        try:
            bound_statement = session.prepare(query)
            __context__['cassandra_cql_prepared'][statement_name] = bound_statement
        except BaseException as e:
            log.critical('Unexpected error while preparing SQL statement: %s', e)
            raise
    else:
        bound_statement = __context__['cassandra_cql_prepared'][statement_name]

    session.row_factory = dict_factory
    ret = []

    try:
        if asynchronous:
            future_results = session.execute_async(bound_statement.bind(statement_arguments))
            # future_results.add_callbacks(_async_log_errors)
        else:
            results = session.execute(bound_statement.bind(statement_arguments))
    except BaseException as e:
        log.error('Failed to execute query: %s\n reason: %s', query, e)
        msg = "ERROR: Cassandra query failed: {0} reason: {1}".format(query, e)
        raise CommandExecutionError(msg)

    if not asynchronous and results:
        for result in results:
            values = {}
            for key, value in six.iteritems(result):
                # Salt won't return dictionaries with odd types like uuid.UUID
                if not isinstance(value, six.text_type):
                    # Must support Cassandra collection types.
                    # Namely, Cassandras set, list, and map collections.
                    if not isinstance(value, (set, list, dict)):
                        value = six.text_type(value)
                values[key] = value
            ret.append(values)

    # If this was a synchronous call, then we either have an empty list
    # because there was no return, or we have a return
    # If this was an asynchronous call we only return the empty list
    return ret