<commit_before>from cassandra.cluster import Cluster, Session

try:
    from cassandra.io.libevreactor import LibevConnection
    ConnectionClass = LibevConnection
except ImportError:
    from cassandra.io.asyncorereactor import AsyncoreConnection
    ConnectionClass = AsyncoreConnection


def _execute_many(self, queries, trace=False):
    """
    Executes many queries in parallel and synchronously waits for the responses.
    """
    futures = (self.execute_async(query, trace=trace) for query in queries)

    for future in futures:
        try:
            yield future.result()
        except Exception:
            yield None

Session.execute_many = _execute_many


class Client(object):
    def __init__(self, servers, **kwargs):
        hosts, port = set(), '9042'
        for server in servers:
            host, port = server.split(':', 1)
            hosts.add(host)

        self._cluster = Cluster(hosts, port=int(port), **kwargs)
        self._cluster.connection_class = ConnectionClass
        self._session = self._cluster.connect()

        self.keyspace = "cache"
        self.column_family = "memcached"

        self._session.set_keyspace(self.keyspace)

        # Prepare all of the necessary statements beforehand
        self._GET = self._session.prepare("SELECT value, flags FROM %s WHERE key = ? LIMIT 1" % self.column_family)
        self._SET = self._session.prepare("INSERT INTO %s (key, value, flags) VALUES (?, ?, ?)" % self.column_family)
        self._DELETE = self._session.prepare("DELETE FROM %s WHERE key = ?" % self.column_family)
        # Cannot be prepared with a dynamic TTL pre C* 2.0<commit_msg>Only patch in execute_many if not found<commit_after>from cassandra.cluster import Cluster, Session

try:
    from cassandra.io.libevreactor import LibevConnection
    ConnectionClass = LibevConnection
except ImportError:
    from cassandra.io.asyncorereactor import AsyncoreConnection
    ConnectionClass = AsyncoreConnection


if not hasattr(Session, 'execute_many'):
    def _execute_many(self, queries, trace=False):
        """
        Executes many queries in parallel and synchronously waits for the responses.
        """
        futures = (self.execute_async(query, trace=trace) for query in queries)

        for future in futures:
            try:
                yield future.result()
            except Exception:
                yield None

    Session.execute_many = _execute_many


class Client(object):
    def __init__(self, servers, **kwargs):
        hosts, port = set(), '9042'
        for server in servers:
            host, port = server.split(':', 1)
            hosts.add(host)

        self._cluster = Cluster(hosts, port=int(port), **kwargs)
        self._cluster.connection_class = ConnectionClass
        self._session = self._cluster.connect()

        self.keyspace = "cache"
        self.column_family = "memcached"

        self._session.set_keyspace(self.keyspace)

        # Prepare all of the necessary statements beforehand
        self._GET = self._session.prepare("SELECT value, flags FROM %s WHERE key = ? LIMIT 1" % self.column_family)
        self._SET = self._session.prepare("INSERT INTO %s (key, value, flags) VALUES (?, ?, ?)" % self.column_family)
        self._DELETE = self._session.prepare("DELETE FROM %s WHERE key = ?" % self.column_family)
        # Cannot be prepared with a dynamic TTL pre C* 2.0
{'fraction_non_alphanumeric': 0.09490126731506042, 'fraction_numerical': 0.004715590922487474, 'mean_word_length': 2.861205915813424, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '169257', 'n_tokens_mistral': 958, 'n_tokens_neox': 914, 'n_words': 273}