Class CassandraSharedSession

join/LSH/datasketch/storage.py:257–304

Cassandra session shared across all storage instances.

if cassandra is not None:

    class CassandraSharedSession(object):
        """Cassandra session shared across all storage instances."""

        __session = None
        __session_buffer = None
        __session_select_buffer = None

        QUERY_CREATE_KEYSPACE = """
            CREATE KEYSPACE IF NOT EXISTS {keyspace}
            WITH replication = {replication}
        """

        QUERY_DROP_KEYSPACE = "DROP KEYSPACE IF EXISTS {}"

        @classmethod
        def get_session(cls, seeds, **kwargs):
            _ = kwargs
            keyspace = kwargs["keyspace"]
            replication = kwargs["replication"]

            if cls.__session is None:
                # Allow dependency injection
                session = kwargs.get("session")
                if session is None:
                    cluster = c_cluster.Cluster(seeds)
                    session = cluster.connect()
                cls.__session = session
            if cls.__session.keyspace != keyspace:
                if kwargs.get("drop_keyspace", False):
                    cls.__session.execute(cls.QUERY_DROP_KEYSPACE.format(keyspace))
                cls.__session.execute(cls.QUERY_CREATE_KEYSPACE.format(
                    keyspace=keyspace,
                    replication=str(replication),
                ))
                cls.__session.set_keyspace(keyspace)
            return cls.__session

        @classmethod
        def get_buffer(cls):
            if cls.__session_buffer is None:
                cls.__session_buffer = []
            return cls.__session_buffer

        @classmethod
        def get_select_buffer(cls):
            if cls.__session_select_buffer is None:
                cls.__session_select_buffer = []
            return cls.__session_select_buffer


    class CassandraClient(object):