<commit_before>import re
import redis

from skew.backends.base import BaseQueue, BaseResultStore


class RedisQueue(BaseQueue):
    """
    A simple Queue that uses the redis to store messages
    """
    def __init__(self, name, connection):
        """
        QUEUE_CONNECTION = 'host:port:database' or defaults to localhost:6379:0
        """
        super(RedisQueue, self).__init__(name, connection)
        
        if not connection:
            connection = 'localhost:6379:0'
        
        self.queue_name = 'skew.redis.%s' % re.sub('[^a-z0-9]', '', name)
        host, port, db = connection.split(':')

        self.conn = redis.Redis(
            host=host, port=int(port), db=int(db)
        )
    
    def write(self, data):
        self.conn.lpush(self.queue_name, data)
    
    def read(self):
        return self.conn.rpop(self.queue_name)
    
    def flush(self):
        self.conn.delete(self.queue_name)
    
    def __len__(self):
        return self.conn.llen(self.queue_name)


class RedisBlockingQueue(RedisQueue):
    """
    Use the blocking right pop, should result in messages getting
    executed close to immediately by the consumer as opposed to
    being polled for
    """
    blocking = True

    def read(self):
        return self.conn.brpop(self.queue_name)


class RedisResultStore(BaseResultStore):
    def __init__(self, name, connection):
        super(RedisResultStore, self).__init__(name, connection)
        
        if not connection:
            connection = 'localhost:6379:0'
        
        self.storage_name = 'skew.redis.results.%s' % re.sub('[^a-z0-9]', '', name)
        host, port, db = connection.split(':')

        self.conn = redis.Redis(
            host=host, port=int(port), db=int(db)
        )
    <commit_msg>Allow passing arbitrary keyword args to the redis connection<commit_after>import re
import redis

from skew.backends.base import BaseQueue, BaseResultStore


class RedisQueue(BaseQueue):
    """
    A simple Queue that uses the redis to store messages
    """
    def __init__(self, name, connection, **connection_kwargs):
        """
        QUEUE_CONNECTION = 'host:port:database' or defaults to localhost:6379:0
        """
        super(RedisQueue, self).__init__(name, connection)
        
        if not connection:
            connection = 'localhost:6379:0'
        
        self.queue_name = 'skew.redis.%s' % re.sub('[^a-z0-9]', '', name)
        host, port, db = connection.split(':')

        self.conn = redis.Redis(
            host=host, port=int(port), db=int(db), **connection_kwargs
        )
    
    def write(self, data):
        self.conn.lpush(self.queue_name, data)
    
    def read(self):
        return self.conn.rpop(self.queue_name)
    
    def flush(self):
        self.conn.delete(self.queue_name)
    
    def __len__(self):
        return self.conn.llen(self.queue_name)


class RedisBlockingQueue(RedisQueue):
    """
    Use the blocking right pop, should result in messages getting
    executed close to immediately by the consumer as opposed to
    being polled for
    """
    blocking = True

    def read(self):
        return self.conn.brpop(self.queue_name)


class RedisResultStore(BaseResultStore):
    def __init__(self, name, connection, **connection_kwargs):
        super(RedisResultStore, self).__init__(name, connection)
        
        if not connection:
            connection = 'localhost:6379:0'
        
        self.storage_name = 'skew.redis.results.%s' % re.sub('[^a-z0-9]', '', name)
        host, port, db = connection.split(':')

        self.conn = redis.Redis(
            host=host, port=int(port), db=int(db), **connection_kwargs
        )
    
{'fraction_non_alphanumeric': 0.1175833107558927, 'fraction_numerical': 0.010295312923327011, 'mean_word_length': 2.4536950420954162, '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': '6935371', 'n_tokens_mistral': 1168, 'n_tokens_neox': 1131, 'n_words': 264}