Redis Emulated Transaction (Python & redis-py)

redis-transaction.py

import redis
from operator import methodcaller

class RedisTransaction(object):
    def __init__(self, **kwargs):
        ''' Transaction timeout sets TTL for copied keys '''
        if 'timeout' in kwargs:
            self.TRANSACTION_TIMEOUT = kwargs['timeout']
        else:
            self.TRANSACTION_TIMEOUT = 10
        transaction_instance = {
            'host' : kwargs['transaction_host'],
            'port' : kwargs['transaction_port'],
            'db'   : kwargs['transaction_db'],
        }
        instance = {
            'host' : kwargs['host'],
            'port' : kwargs['port'],
            'db'   : kwargs['db'],
        }
        self.USE_SCRIPTING = instance == transaction_instance
        self.R = redis.StrictRedis(**instance)
        self.R_T = redis.StrictRedis(**transaction_instance)
        
    def begin(self, *args):
        transaction_params = {
            'timeout' : self.TRANSACTION_TIMEOUT,
        }
        ''' Start by copying keys '''
        '''
        Redis Lua Scripting
        EVAL "return redis.call('RESTORE', ARGV[1] .. ':' .. KEYS[1], 0, redis.call('DUMP', KEYS[1]))" 1 a1 prefix
        '''
        script = '''
        local k = 1
        for k = 1, tonumber(ARGV[1]) do
          redis.call('RESTORE', ARGV[2] .. ':' .. KEYS[k], %(timeout)s, redis.call('DUMP', KEYS[k]))
        end
        return redis.status_reply("1")
        ''' % transaction_params
        keys = dict([ (arg, 1) for arg in args ]).keys()
        self.TRANSACTION_ID = self.R_T.incr('GTID')
        self.TRANSACTION_KEYS = map(self.keymaker, keys)
        if self.USE_SCRIPTING:
            arguments = [ len(keys), self.TRANSACTION_ID ]
            arguments.extend(keys[:])
            if not self.R.eval(script, len(keys), *arguments):
                return False
        else:
            self.TRANSACTION_KEYS = []
            for key in args:
                try:
                  self.R_T.restore(self.keymaker(transaction_id, key) self.TRANSACTION_TIMEOUT, r.dump(key))
                except:  
                  return False
        return self.TRANSACTION_KEYS
    
    def rcommand(method, keys, *args, **kwargs):
        ''' Generic Redis command caller '''
        if isinstance(keys, list):
           keys = self.keymaker(keys)
        else:
            keys = map(self.keymaker, keys)
        ''' need to check what parameters the redis-py API methods have '''
        C = methodcaller(method, keys, *args, **kwargs)
        return C(self.R_T)
    
    def commit(self):
        ''' Commit the pseudo-transaction with MULTI-EXEC'''
        P = self.R.pipeline()
        for key in self.TRANSACTION_KEYS:
            P.restore(key, self.R_T.ttl(key), self.R_T.dump(key))
        P.execute()
        self.TRANSACTION_KEYS = []
    
    def rollback(self):
        ''' Rollback simply deletes every transaction key '''
        self.R_T.delete(*self.TRANSACTION_KEYS)
    
    def cleanup(self):
        self.rollback()
    
    def keymaker(self, key):
        ''' Add transaction namespace to a key '''
        return "%d:%s" % (self.TRANSACTION_ID, key)
    

Possible to add configurable isolation levels?

• READ-UNCOMMITTED -> "dirty" reads(requires keeping track of other active transactions)
• READ-COMMITTED -> only committed values will be read (default)
• REPEATABLE-READ -> read values are frozen to their initially read values (snapshot/consistent reads)