tydmiMcd

1import datetime
2import errno
3import socket
4import threading
5import time
6import warnings
7from itertools import chain, imap
8from redis.exceptions import ConnectionError, ResponseError, InvalidResponse, WatchError
9from redis.exceptions import RedisError, AuthenticationError
10
11
12class ConnectionPool(threading.local):
13    "Manages a list of connections on the local thread"
14    def __init__(self):
15        self.connections = {}
16
17    def make_connection_key(self, host, port, db):
18        "Create a unique key for the specified host, port and db"
19        return '%s:%s:%s' % (host, port, db)
20
21    def get_connection(self, host, port, db, password, socket_timeout):
22        "Return a specific connection for the specified host, port and db"
23        key = self.make_connection_key(host, port, db)
24        if key not in self.connections:
25            self.connections[key] = Connection(
26                host, port, db, password, socket_timeout)
27        return self.connections[key]
28
29    def get_all_connections(self):
30        "Return a list of all connection objects the manager knows about"
31        return self.connections.values()
32
33
34class Connection(object):
35    "Manages TCP communication to and from a Redis server"
36    def __init__(self, host='localhost', port=6379, db=0, password=None,
37                 socket_timeout=None):
38        self.host = host
39        self.port = port
40        self.db = db
41        self.password = password
42        self.socket_timeout = socket_timeout
43        self._sock = None
44        self._fp = None
45
46    def connect(self, redis_instance):
47        "Connects to the Redis server if not already connected"
48        if self._sock:
49            return
50        try:
51            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
52            sock.settimeout(self.socket_timeout)
53            sock.connect((self.host, self.port))
54        except socket.error, e:
55            # args for socket.error can either be (errno, "message")
56            # or just "message"
57            if len(e.args) == 1:
58                error_message = "Error connecting to %s:%s. %s." % \
59                    (self.host, self.port, e.args[0])
60            else:
61                error_message = "Error %s connecting %s:%s. %s." % \
62                    (e.args[0], self.host, self.port, e.args[1])
63            raise ConnectionError(error_message)
64        sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
65        self._sock = sock
66        self._fp = sock.makefile('r')
67        redis_instance._setup_connection()
68
69    def disconnect(self):
70        "Disconnects from the Redis server"
71        if self._sock is None:
72            return
73        try:
74            self._sock.close()
75        except socket.error:
76            pass
77        self._sock = None
78        self._fp = None
79
80    def send(self, command, redis_instance):
81        "Send ``command`` to the Redis server. Return the result."
82        self.connect(redis_instance)
83        try:
84            self._sock.sendall(command)
85        except socket.error, e:
86            if e.args[0] == errno.EPIPE:
87                self.disconnect()
88            if isinstance(e.args, basestring):
89                errno, errmsg = 'UNKNOWN', e.args
90            else:
91                errno, errmsg = e.args
92            raise ConnectionError("Error %s while writing to socket. %s." % \
93                (errno, errmsg))
94
95    def read(self, length=None):
96        """
97        Read a line from the socket is length is None,
98        otherwise read ``length`` bytes
99        """
100        try:
101            if length is not None:
102                return self._fp.read(length)
103            return self._fp.readline()
104        except socket.error, e:
105            self.disconnect()
106            if e.args and e.args[0] == errno.EAGAIN:
107                raise ConnectionError("Error while reading from socket: %s" % \
108                    e.args[1])
109        return ''
110
111def list_or_args(command, keys, args):
112    # returns a single list combining keys and args
113    # if keys is not a list or args has items, issue a
114    # deprecation warning
115    oldapi = bool(args)
116    try:
117        i = iter(keys)
118        # a string can be iterated, but indicates
119        # keys wasn't passed as a list
120        if isinstance(keys, basestring):
121            oldapi = True
122    except TypeError:
123        oldapi = True
124        keys = [keys]
125    if oldapi:
126        warnings.warn(DeprecationWarning(
127            "Passing *args to Redis.%s has been deprecated. "
128            "Pass an iterable to ``keys`` instead" % command
129        ))
130        keys.extend(args)
131    return keys
132
133def timestamp_to_datetime(response):
134    "Converts a unix timestamp to a Python datetime object"
135    if not response:
136        return None
137    try:
138        response = int(response)
139    except ValueError:
140        return None
141    return datetime.datetime.fromtimestamp(response)
142
143def string_keys_to_dict(key_string, callback):
144    return dict([(key, callback) for key in key_string.split()])
145
146def dict_merge(*dicts):
147    merged = {}
148    [merged.update(d) for d in dicts]
149    return merged
150
151def parse_info(response):
152    "Parse the result of Redis's INFO command into a Python dict"
153    info = {}
154    def get_value(value):
155        if ',' not in value:
156            return value
157        sub_dict = {}
158        for item in value.split(','):
159            k, v = item.split('=')
160            try:
161                sub_dict[k] = int(v)
162            except ValueError:
163                sub_dict[k] = v
164        return sub_dict
165    for line in response.splitlines():
166        key, value = line.split(':')
167        try:
168            info[key] = int(value)
169        except ValueError:
170            info[key] = get_value(value)
171    return info
172
173def pairs_to_dict(response):
174    "Create a dict given a list of key/value pairs"
175    return dict(zip(response[::2], response[1::2]))
176
177def zset_score_pairs(response, **options):
178    """
179    If ``withscores`` is specified in the options, return the response as
180    a list of (value, score) pairs
181    """
182    if not response or not options['withscores']:
183        return response
184    return zip(response[::2], map(float, response[1::2]))
185
186def int_or_none(response):
187    if response is None:
188        return None
189    return int(response)
190
191def float_or_none(response):
192    if response is None:
193        return None
194    return float(response)
195
196def parse_config(response, **options):
197    # this is stupid, but don't have a better option right now
198    if options['parse'] == 'GET':
199        return response and pairs_to_dict(response) or {}
200    return response == 'OK'
201
202class Redis(threading.local):
203    """
204    Implementation of the Redis protocol.
205
206    This abstract class provides a Python interface to all Redis commands
207    and an implementation of the Redis protocol.
208
209    Connection and Pipeline derive from this, implementing how
210    the commands are sent and received to the Redis server
211    """
212    RESPONSE_CALLBACKS = dict_merge(
213        string_keys_to_dict(
214            'AUTH DEL EXISTS EXPIRE EXPIREAT HDEL HEXISTS HMSET MOVE MSETNX '
215            'PERSIST RENAMENX SADD SISMEMBER SMOVE SETEX SETNX SREM ZADD ZREM'
216            ### ALCHEMY DATABASE ###
217            'CREATE DROP INSERT',
218            bool
219            ),
220        string_keys_to_dict(
221            'DECRBY GETBIT HLEN INCRBY LINSERT LLEN LPUSHX RPUSHX SCARD '
222            'SDIFFSTORE SETBIT SETRANGE SINTERSTORE STRLEN SUNIONSTORE ZCARD '
223            'ZREMRANGEBYRANK ZREMRANGEBYSCORE',
224            int
225            ),
226        string_keys_to_dict(
227            # these return OK, or int if redis-server is >=1.3.4
228            'LPUSH RPUSH',
229            lambda r: isinstance(r, int) and r or r == 'OK'
230            ),
231        string_keys_to_dict('ZSCORE ZINCRBY', float_or_none),
232        string_keys_to_dict(
233            'FLUSHALL FLUSHDB LSET LTRIM MSET RENAME '
234            ###'SAVE SELECT SET SHUTDOWN SLAVEOF WATCH UNWATCH',
235            ### ALCHEMY DATABASE ###
236            'SAVE SET SHUTDOWN SLAVEOF WATCH UNWATCH',
237            lambda r: r == 'OK'
238            ),
239        string_keys_to_dict('BLPOP BRPOP', lambda r: r and tuple(r) or None),
240        string_keys_to_dict('SDIFF SINTER SMEMBERS SUNION',
241            lambda r: r and set(r) or set()
242            ),
243        string_keys_to_dict('ZRANGE ZRANGEBYSCORE ZREVRANGE', zset_score_pairs),
244        string_keys_to_dict('ZRANK ZREVRANK', int_or_none),
245        {
246            'BGREWRITEAOF': lambda r: \
247                r == 'Background rewriting of AOF file started',
248            'BGSAVE': lambda r: r == 'Background saving started',
249            'BRPOPLPUSH': lambda r: r and r or None,
250            'CONFIG': parse_config,
251            'HGETALL': lambda r: r and pairs_to_dict(r) or {},
252            'INFO': parse_info,
253            'LASTSAVE': timestamp_to_datetime,
254            'PING': lambda r: r == 'PONG',
255            'RANDOMKEY': lambda r: r and r or None,
256            'TTL': lambda r: r != -1 and r or None,
257        }
258        )
259
260    # commands that should NOT pull data off the network buffer when executed
261    SUBSCRIPTION_COMMANDS = set([
262        'SUBSCRIBE', 'UNSUBSCRIBE', 'PSUBSCRIBE', 'PUNSUBSCRIBE'
263        ])
264
265    def __init__(self, host='localhost', port=6379,
266                 db=0, password=None, socket_timeout=None,
267                 connection_pool=None,
268                 charset='utf-8', errors='strict'):
269        self.encoding = charset
270        self.errors = errors
271        self.connection = None
272        self.subscribed = False
273        self.connection_pool = connection_pool and connection_pool or ConnectionPool()
274        self.select(db, host, port, password, socket_timeout)
275
276    #### Legacty accessors of connection information ####
277    def _get_host(self):
278        return self.connection.host
279    host = property(_get_host)
280
281    def _get_port(self):
282        return self.connection.port
283    port = property(_get_port)
284
285    def _get_db(self):
286        return self.connection.db
287    db = property(_get_db)
288
289    def pipeline(self, transaction=True):
290        """
291        Return a new pipeline object that can queue multiple commands for
292        later execution. ``transaction`` indicates whether all commands
293        should be executed atomically. Apart from multiple atomic operations,
294        pipelines are useful for batch loading of data as they reduce the
295        number of back and forth network operations between client and server.
296        """
297        return Pipeline(
298            self.connection,
299            transaction,
300            self.encoding,
301            self.errors
302            )
303
304    def lock(self, name, timeout=None, sleep=0.1):
305        """
306        Return a new Lock object using key ``name`` that mimics
307        the behavior of threading.Lock.
308
309        If specified, ``timeout`` indicates a maximum life for the lock.
310        By default, it will remain locked until release() is called.
311
312        ``sleep`` indicates the amount of time to sleep per loop iteration
313        when the lock is in blocking mode and another client is currently
314        holding the lock.
315        """
316        return Lock(self, name, timeout=timeout, sleep=sleep)
317
318    #### COMMAND EXECUTION AND PROTOCOL PARSING ####
319    def _execute_command(self, command_name, command, **options):
320        subscription_command = command_name in self.SUBSCRIPTION_COMMANDS
321        if self.subscribed and not subscription_command:
322            raise RedisError("Cannot issue commands other than SUBSCRIBE and "
323                "UNSUBSCRIBE while channels are open")
324        try:
325            self.connection.send(command, self)
326            if subscription_command:
327                return None
328            return self.parse_response(command_name, **options)
329        except ConnectionError:
330            self.connection.disconnect()
331            self.connection.send(command, self)
332            if subscription_command:
333                return None
334            return self.parse_response(command_name, **options)
335
336    def execute_command(self, *args, **options):
337        "Sends the command to the redis server and returns it's response"
338        cmds = ['$%s\r\n%s\r\n' % (len(enc_value), enc_value)
339                for enc_value in imap(self.encode, args)]
340        return self._execute_command(
341            args[0],
342            '*%s\r\n%s' % (len(cmds), ''.join(cmds)),
343            **options
344            )
345
346    def _parse_response(self, command_name, catch_errors):
347        conn = self.connection
348        response = conn.read()[:-2] # strip last two characters (\r\n)
349        if not response:
350            self.connection.disconnect()
351            raise ConnectionError("Socket closed on remote end")
352
353        # server returned a null value
354        if response in ('$-1', '*-1'):
355            return None
356        byte, response = response[0], response[1:]
357
358        # server returned an error
359        if byte == '-':
360            if response.startswith('ERR '):
361                response = response[4:]
362            raise ResponseError(response)
363        # single value
364        elif byte == '+':
365            return response
366        # int value
367        elif byte == ':':
368            return int(response)
369        # bulk response
370        elif byte == '$':
371            length = int(response)
372            if length == -1:
373                return None
374            response = length and conn.read(length) or ''
375            conn.read(2) # read the \r\n delimiter
376            return response
377        # multi-bulk response
378        elif byte == '*':
379            length = int(response)
380            if length == -1:
381                return None
382            if not catch_errors:
383                return [self._parse_response(command_name, catch_errors)
384                    for i in range(length)]
385            else:
386                # for pipelines, we need to read everything,
387                # including response errors. otherwise we'd
388                # completely mess up the receive buffer
389                data = []
390                for i in range(length):
391                    try:
392                        data.append(
393                            self._parse_response(command_name, catch_errors)
394                            )
395                    except Exception, e:
396                        data.append(e)
397                return data
398
399        raise InvalidResponse("Unknown response type for: %s" % command_name)
400
401    def parse_response(self, command_name, catch_errors=False, **options):
402        "Parses a response from the Redis server"
403        response = self._parse_response(command_name, catch_errors)
404        if command_name in self.RESPONSE_CALLBACKS:
405            return self.RESPONSE_CALLBACKS[command_name](response, **options)
406        return response
407
408    def encode(self, value):
409        "Encode ``value`` using the instance's charset"
410        if isinstance(value, str):
411            return value
412        if isinstance(value, unicode):
413            return value.encode(self.encoding, self.errors)
414        # not a string or unicode, attempt to convert to a string
415        return str(value)
416
417    #### CONNECTION HANDLING ####
418    def get_connection(self, host, port, db, password, socket_timeout):
419        "Returns a connection object"
420        conn = self.connection_pool.get_connection(
421            host, port, db, password, socket_timeout)
422        # if for whatever reason the connection gets a bad password, make
423        # sure a subsequent attempt with the right password makes its way
424        # to the connection
425        conn.password = password
426        return conn
427
428    def _setup_connection(self):
429        """
430        After successfully opening a socket to the Redis server, the
431        connection object calls this method to authenticate and select
432        the appropriate database.
433        """
434        if self.connection.password:
435            if not self.execute_command('AUTH', self.connection.password):
436                raise AuthenticationError("Invalid Password")
437        self.execute_command('SELECT', self.connection.db)
438
439    def select(self, db, host=None, port=None, password=None,
440            socket_timeout=None):
441        """
442        Switch to a different Redis connection.
443
444        If the host and port aren't provided and there's an existing
445        connection, use the existing connection's host and port instead.
446
447        Note this method actually replaces the underlying connection object
448        prior to issuing the SELECT command.  This makes sure we protect
449        the thread-safe connections
450        """
451        if host is None:
452            if self.connection is None:
453                raise RedisError("A valid hostname or IP address "
454                    "must be specified")
455            host = self.connection.host
456        if port is None:
457            if self.connection is None:
458                raise RedisError("A valid port must be specified")
459            port = self.connection.port
460
461        self.connection = self.get_connection(
462            host, port, db, password, socket_timeout)
463
464    def shutdown(self):
465        "Shutdown the server"
466        if self.subscribed:
467            raise RedisError("Can't call 'shutdown' from a pipeline'")
468        try:
469            self.execute_command('SHUTDOWN')
470        except ConnectionError:
471            # a ConnectionError here is expected
472            return
473        raise RedisError("SHUTDOWN seems to have failed.")
474
475
476    #### SERVER INFORMATION ####
477    def bgrewriteaof(self):
478        "Tell the Redis server to rewrite the AOF file from data in memory."
479        return self.execute_command('BGREWRITEAOF')
480
481    def bgsave(self):
482        """
483        Tell the Redis server to save its data to disk.  Unlike save(),
484        this method is asynchronous and returns immediately.
485        """
486        return self.execute_command('BGSAVE')
487
488    def config_get(self, pattern="*"):
489        "Return a dictionary of configuration based on the ``pattern``"
490        return self.execute_command('CONFIG', 'GET', pattern, parse='GET')
491
492    def config_set(self, name, value):
493        "Set config item ``name`` with ``value``"
494        return self.execute_command('CONFIG', 'SET', name, value, parse='SET')
495
496    def dbsize(self):
497        "Returns the number of keys in the current database"
498        return self.execute_command('DBSIZE')
499
500    def delete(self, *names):
501        "Delete one or more keys specified by ``names``"
502        return self.execute_command('DEL', *names)
503    __delitem__ = delete
504
505    def flush(self, all_dbs=False):
506        warnings.warn(DeprecationWarning(
507            "'flush' has been deprecated. "
508            "Use Redis.flushdb() or Redis.flushall() instead"))
509        if all_dbs:
510            return self.flushall()
511        return self.flushdb()
512
513    def flushall(self):
514        "Delete all keys in all databases on the current host"
515        return self.execute_command('FLUSHALL')
516
517    def flushdb(self):
518        "Delete all keys in the current database"
519        return self.execute_command('FLUSHDB')
520
521    def info(self):
522        "Returns a dictionary containing information about the Redis server"
523        return self.execute_command('INFO')
524
525    def lastsave(self):
526        """
527        Return a Python datetime object representing the last time the
528        Redis database was saved to disk
529        """
530        return self.execute_command('LASTSAVE')
531
532    def ping(self):
533        "Ping the Redis server"
534        return self.execute_command('PING')
535
536    def save(self):
537        """
538        Tell the Redis server to save its data to disk,
539        blocking until the save is complete
540        """
541        return self.execute_command('SAVE')
542
543    def slaveof(self, host=None, port=None):
544        """
545        Set the server to be a replicated slave of the instance identified
546        by the ``host`` and ``port``. If called without arguements, the
547        instance is promoted to a master instead.
548        """
549        if host is None and port is None:
550            return self.execute_command("SLAVEOF NO ONE")
551        return self.execute_command("SLAVEOF", host, port)
552
553    #### BASIC KEY COMMANDS ####
554    def append(self, key, value):
555        """
556        Appends the string ``value`` to the value at ``key``. If ``key``
557        doesn't already exist, create it with a value of ``value``.
558        Returns the new length of the value at ``key``.
559        """
560        return self.execute_command('APPEND', key, value)
561
562    def decr(self, name, amount=1):
563        """
564        Decrements the value of ``key`` by ``amount``.  If no key exists,
565        the value will be initialized as 0 - ``amount``
566        """
567        return self.execute_command('DECRBY', name, amount)
568
569    def exists(self, name):
570        "Returns a boolean indicating whether key ``name`` exists"
571        return self.execute_command('EXISTS', name)
572    __contains__ = exists
573
574    def expire(self, name, time):
575        "Set an expire flag on key ``name`` for ``time`` seconds"
576        return self.execute_command('EXPIRE', name, time)
577
578    def expireat(self, name, when):
579        """
580        Set an expire flag on key ``name``. ``when`` can be represented
581        as an integer indicating unix time or a Python datetime object.
582        """
583        if isinstance(when, datetime.datetime):
584            when = int(time.mktime(when.timetuple()))
585        return self.execute_command('EXPIREAT', name, when)
586
587    def get(self, name):
588        """
589        Return the value at key ``name``, or None of the key doesn't exist
590        """
591        return self.execute_command('GET', name)
592    __getitem__ = get
593
594    def getbit(self, name, offset):
595        "Returns a boolean indicating the value of ``offset`` in ``name``"
596        return self.execute_command('GETBIT', name, offset)
597
598    def getset(self, name, value):
599        """
600        Set the value at key ``name`` to ``value`` if key doesn't exist
601        Return the value at key ``name`` atomically
602        """
603        return self.execute_command('GETSET', name, value)
604
605    def incr(self, name, amount=1):
606        """
607        Increments the value of ``key`` by ``amount``.  If no key exists,
608        the value will be initialized as ``amount``
609        """
610        return self.execute_command('INCRBY', name, amount)
611
612    def keys(self, pattern='*'):
613        "Returns a list of keys matching ``pattern``"
614        return self.execute_command('KEYS', pattern)
615
616    def mget(self, keys, *args):
617        """
618        Returns a list of values ordered identically to ``keys``
619
620        * Passing *args to this method has been deprecated *
621        """
622        keys = list_or_args('mget', keys, args)
623        return self.execute_command('MGET', *keys)
624
625    def mset(self, mapping):
626        "Sets each key in the ``mapping`` dict to its corresponding value"
627        items = []
628        for pair in mapping.iteritems():
629            items.extend(pair)
630        return self.execute_command('MSET', *items)
631
632    def msetnx(self, mapping):
633        """
634        Sets each key in the ``mapping`` dict to its corresponding value if
635        none of the keys are already set
636        """
637        items = []
638        for pair in mapping.iteritems():
639            items.extend(pair)
640        return self.execute_command('MSETNX', *items)
641
642    def move(self, name, db):
643        "Moves the key ``name`` to a different Redis database ``db``"
644        return self.execute_command('MOVE', name, db)
645
646    def persist(self, name):
647        "Removes an expiration on ``name``"
648        return self.execute_command('PERSIST', name)
649
650    def randomkey(self):
651        "Returns the name of a random key"
652        return self.execute_command('RANDOMKEY')
653
654    def rename(self, src, dst, **kwargs):
655        """
656        Rename key ``src`` to ``dst``
657
658        * The following flags have been deprecated *
659        If ``preserve`` is True, rename the key only if the destination name
660            doesn't already exist
661        """
662        if kwargs:
663            if 'preserve' in kwargs:
664                warnings.warn(DeprecationWarning(
665                    "preserve option to 'rename' is deprecated, "
666                    "use Redis.renamenx instead"))
667                if kwargs['preserve']:
668                    return self.renamenx(src, dst)
669        return self.execute_command('RENAME', src, dst)
670
671    def renamenx(self, src, dst):
672        "Rename key ``src`` to ``dst`` if ``dst`` doesn't already exist"
673        return self.execute_command('RENAMENX', src, dst)
674
675
676    def set(self, name, value, **kwargs):
677        """
678        Set the value at key ``name`` to ``value``
679
680        * The following flags have been deprecated *
681        If ``preserve`` is True, set the value only if key doesn't already
682        exist
683        If ``getset`` is True, set the value only if key doesn't already exist
684        and return the resulting value of key
685        """
686        if kwargs:
687            if 'getset' in kwargs:
688                warnings.warn(DeprecationWarning(
689                    "getset option to 'set' is deprecated, "
690                    "use Redis.getset() instead"))
691                if kwargs['getset']:
692                    return self.getset(name, value)
693            if 'preserve' in kwargs:
694                warnings.warn(DeprecationWarning(
695                    "preserve option to 'set' is deprecated, "
696                    "use Redis.setnx() instead"))
697                if kwargs['preserve']:
698                    return self.setnx(name, value)
699        return self.execute_command('SET', name, value)
700    __setitem__ = set
701
702    def setbit(self, name, offset, value):
703        """
704        Flag the ``offset`` in ``name`` as ``value``. Returns a boolean
705        indicating the previous value of ``offset``.
706        """
707        value = value and 1 or 0
708        return self.execute_command('SETBIT', name, offset, value)
709
710    def setex(self, name, value, time):
711        """
712        Set the value of key ``name`` to ``value``
713        that expires in ``time`` seconds
714        """
715        return self.execute_command('SETEX', name, time, value)
716
717    def setnx(self, name, value):
718        "Set the value of key ``name`` to ``value`` if key doesn't exist"
719        return self.execute_command('SETNX', name, value)
720
721    def setrange(self, name, offset, value):
722        """
723        Overwrite bytes in the value of ``name`` starting at ``offset`` with
724        ``value``. If ``offset`` plus the length of ``value`` exceeds the
725        length of the original value, the new value will be larger than before.
726        If ``offset`` exceeds the length of the original value, null bytes
727        will be used to pad between the end of the previous value and the start
728        of what's being injected.
729
730        Returns the length of the new string.
731        """
732        return self.execute_command('SETRANGE', name, offset, value)
733
734    def strlen(self, name):
735        "Return the number of bytes stored in the value of ``name``"
736        return self.execute_command('STRLEN', name)
737
738    def substr(self, name, start, end=-1):
739        """
740        Return a substring of the string at key ``name``. ``start`` and ``end``
741        are 0-based integers specifying the portion of the string to return.
742        """
743        return self.execute_command('SUBSTR', name, start, end)
744
745    def ttl(self, name):
746        "Returns the number of seconds until the key ``name`` will expire"
747        return self.execute_command('TTL', name)
748
749    def type(self, name):
750        "Returns the type of key ``name``"
751        return self.execute_command('TYPE', name)
752
753    def watch(self, name):
754        """
755        Watches the value at key ``name``, or None of the key doesn't exist
756        """
757        if self.subscribed:
758            raise RedisError("Can't call 'watch' from a pipeline'")
759
760        return self.execute_command('WATCH', name)
761
762    def unwatch(self):
763        """
764        Unwatches the value at key ``name``, or None of the key doesn't exist
765        """
766        if self.subscribed:
767            raise RedisError("Can't call 'unwatch' from a pipeline'")
768
769        return self.execute_command('UNWATCH')
770
771    #### LIST COMMANDS ####
772    def blpop(self, keys, timeout=0):
773        """
774        LPOP a value off of the first non-empty list
775        named in the ``keys`` list.
776
777        If none of the lists in ``keys`` has a value to LPOP, then block
778        for ``timeout`` seconds, or until a value gets pushed on to one
779        of the lists.
780
781        If timeout is 0, then block indefinitely.
782        """
783        if timeout is None:
784            timeout = 0
785        if isinstance(keys, basestring):
786            keys = [keys]
787        else:
788            keys = list(keys)
789        keys.append(timeout)
790        return self.execute_command('BLPOP', *keys)
791
792    def brpop(self, keys, timeout=0):
793        """
794        RPOP a value off of the first non-empty list
795        named in the ``keys`` list.
796
797        If none of the lists in ``keys`` has a value to LPOP, then block
798        for ``timeout`` seconds, or until a value gets pushed on to one
799        of the lists.
800
801        If timeout is 0, then block indefinitely.
802        """
803        if timeout is None:
804            timeout = 0
805        if isinstance(keys, basestring):
806            keys = [keys]
807        else:
808            keys = list(keys)
809        keys.append(timeout)
810        return self.execute_command('BRPOP', *keys)
811
812    def brpoplpush(self, src, dst, timeout=0):
813        """
814        Pop a value off the tail of ``src``, push it on the head of ``dst``
815        and then return it.
816
817        This command blocks until a value is in ``src`` or until ``timeout``
818        seconds elapse, whichever is first. A ``timeout`` value of 0 blocks
819        forever.
820        """
821        if timeout is None:
822            timeout = 0
823        return self.execute_command('BRPOPLPUSH', src, dst, timeout)
824
825    def lindex(self, name, index):
826        """
827        Return the item from list ``name`` at position ``index``
828
829        Negative indexes are supported and will return an item at the
830        end of the list
831        """
832        return self.execute_command('LINDEX', name, index)
833    
834    def linsert(self, name, where, refvalue, value):
835        """
836        Insert ``value`` in list ``name`` either immediately before or after
837        [``where``] ``refvalue``
838        
839        Returns the new length of the list on success or -1 if ``refvalue``
840        is not in the list.
841        """
842        return self.execute_command('LINSERT', name, where, refvalue, value)
843    
844    def llen(self, name):
845        "Return the length of the list ``name``"
846        return self.execute_command('LLEN', name)
847
848    def lpop(self, name):
849        "Remove and return the first item of the list ``name``"
850        return self.execute_command('LPOP', name)
851
852    def lpush(self, name, value):
853        "Push ``value`` onto the head of the list ``name``"
854        return self.execute_command('LPUSH', name, value)
855    
856    def lpushx(self, name, value):
857        "Push ``value`` onto the head of the list ``name`` if ``name`` exists"
858        return self.execute_command('LPUSHX', name, value)
859
860    def lrange(self, name, start, end):
861        """
862        Return a slice of the list ``name`` between
863        position ``start`` and ``end``
864
865        ``start`` and ``end`` can be negative numbers just like
866        Python slicing notation
867        """
868        return self.execute_command('LRANGE', name, start, end)
869
870    def lrem(self, name, value, num=0):
871        """
872        Remove the first ``num`` occurrences of ``value`` from list ``name``
873
874        If ``num`` is 0, then all occurrences will be removed
875        """
876        return self.execute_command('LREM', name, num, value)
877
878    def lset(self, name, index, value):
879        "Set ``position`` of list ``name`` to ``value``"
880        return self.execute_command('LSET', name, index, value)
881
882    def ltrim(self, name, start, end):
883        """
884        Trim the list ``name``, removing all values not within the slice
885        between ``start`` and ``end``
886
887        ``start`` and ``end`` can be negative numbers just like
888        Python slicing notation
889        """
890        return self.execute_command('LTRIM', name, start, end)
891
892    def pop(self, name, tail=False):
893        """
894        Pop and return the first or last element of list ``name``
895
896        * This method has been deprecated,
897          use Redis.lpop or Redis.rpop instead *
898        """
899        warnings.warn(DeprecationWarning(
900            "Redis.pop has been deprecated, "
901            "use Redis.lpop or Redis.rpop instead"))
902        if tail:
903            return self.rpop(name)
904        return self.lpop(name)
905
906    def push(self, name, value, head=False):
907        """
908        Push ``value`` onto list ``name``.
909
910        * This method has been deprecated,
911          use Redis.lpush or Redis.rpush instead *
912        """
913        warnings.warn(DeprecationWarning(
914            "Redis.push has been deprecated, "
915            "use Redis.lpush or Redis.rpush instead"))
916        if head:
917            return self.lpush(name, value)
918        return self.rpush(name, value)
919
920    def rpop(self, name):
921        "Remove and return the last item of the list ``name``"
922        return self.execute_command('RPOP', name)
923
924    def rpoplpush(self, src, dst):
925        """
926        RPOP a value off of the ``src`` list and atomically LPUSH it
927        on to the ``dst`` list.  Returns the value.
928        """
929        return self.execute_command('RPOPLPUSH', src, dst)
930
931    def rpush(self, name, value):
932        "Push ``value`` onto the tail of the list ``name``"
933        return self.execute_command('RPUSH', name, value)
934
935    def rpushx(self, name, value):
936        "Push ``value`` onto the tail of the list ``name`` if ``name`` exists"
937        return self.execute_command('RPUSHX', name, value)
938
939    def sort(self, name, start=None, num=None, by=None, get=None,
940             desc=False, alpha=False, store=None):
941        """
942        Sort and return the list, set or sorted set at ``name``.
943
944        ``start`` and ``num`` allow for paging through the sorted data
945
946        ``by`` allows using an external key to weight and sort the items.
947            Use an "*" to indicate where in the key the item value is located
948
949        ``get`` allows for returning items from external keys rather than the
950            sorted data itself.  Use an "*" to indicate where int he key
951            the item value is located
952
953        ``desc`` allows for reversing the sort
954
955        ``alpha`` allows for sorting lexicographically rather than numerically
956
957        ``store`` allows for storing the result of the sort into
958            the key ``store``
959        """
960        if (start is not None and num is None) or \
961                (num is not None and start is None):
962            raise RedisError("``start`` and ``num`` must both be specified")
963
964        pieces = [name]
965        if by is not None:
966            pieces.append('BY')
967            pieces.append(by)
968        if start is not None and num is not None:
969            pieces.append('LIMIT')
970            pieces.append(start)
971            pieces.append(num)
972        if get is not None:
973            # If get is a string assume we want to get a single value.
974            # Otherwise assume it's an interable and we want to get multiple
975            # values. We can't just iterate blindly because strings are
976            # iterable.
977            if isinstance(get, basestring):
978                pieces.append('GET')
979                pieces.append(get)
980            else:
981                for g in get:
982                    pieces.append('GET')
983                    pieces.append(g)
984        if desc:
985            pieces.append('DESC')
986        if alpha:
987            pieces.append('ALPHA')
988        if store is not None:
989            pieces.append('STORE')
990            pieces.append(store)
991        return self.execute_command('SORT', *pieces)
992
993
994    #### SET COMMANDS ####
995    def sadd(self, name, value):
996        "Add ``value`` to set ``name``"
997        return self.execute_command('SADD', name, value)
998
999    def scard(self, name):
1000        "Return the number of elements in set ``name``"
1001        return self.execute_command('SCARD', name)
1002
1003    def sdiff(self, keys, *args):
1004        "Return the difference of sets specified by ``keys``"
1005        keys = list_or_args('sdiff', keys, args)
1006        return self.execute_command('SDIFF', *keys)
1007
1008    def sdiffstore(self, dest, keys, *args):
1009        """
1010        Store the difference of sets specified by ``keys`` into a new
1011        set named ``dest``.  Returns the number of keys in the new set.
1012        """
1013        keys = list_or_args('sdiffstore', keys, args)
1014        return self.execute_command('SDIFFSTORE', dest, *keys)
1015
1016    def sinter(self, keys, *args):
1017        "Return the intersection of sets specified by ``keys``"
1018        keys = list_or_args('sinter', keys, args)
1019        return self.execute_command('SINTER', *keys)
1020
1021    def sinterstore(self, dest, keys, *args):
1022        """
1023        Store the intersection of sets specified by ``keys`` into a new
1024        set named ``dest``.  Returns the number of keys in the new set.
1025        """
1026        keys = list_or_args('sinterstore', keys, args)
1027        return self.execute_command('SINTERSTORE', dest, *keys)
1028
1029    def sismember(self, name, value):
1030        "Return a boolean indicating if ``value`` is a member of set ``name``"
1031        return self.execute_command('SISMEMBER', name, value)
1032
1033    def smembers(self, name):
1034        "Return all members of the set ``name``"
1035        return self.execute_command('SMEMBERS', name)
1036
1037    def smove(self, src, dst, value):
1038        "Move ``value`` from set ``src`` to set ``dst`` atomically"
1039        return self.execute_command('SMOVE', src, dst, value)
1040
1041    def spop(self, name):
1042        "Remove and return a random member of set ``name``"
1043        return self.execute_command('SPOP', name)
1044
1045    def srandmember(self, name):
1046        "Return a random member of set ``name``"
1047        return self.execute_command('SRANDMEMBER', name)
1048
1049    def srem(self, name, value):
1050        "Remove ``value`` from set ``name``"
1051        return self.execute_command('SREM', name, value)
1052
1053    def sunion(self, keys, *args):
1054        "Return the union of sets specifiued by ``keys``"
1055        keys = list_or_args('sunion', keys, args)
1056        return self.execute_command('SUNION', *keys)
1057
1058    def sunionstore(self, dest, keys, *args):
1059        """
1060        Store the union of sets specified by ``keys`` into a new
1061        set named ``dest``.  Returns the number of keys in the new set.
1062        """
1063        keys = list_or_args('sunionstore', keys, args)
1064        return self.execute_command('SUNIONSTORE', dest, *keys)
1065
1066
1067    #### SORTED SET COMMANDS ####
1068    def zadd(self, name, value, score):
1069        "Add member ``value`` with score ``score`` to sorted set ``name``"
1070        return self.execute_command('ZADD', name, score, value)
1071
1072    def zcard(self, name):
1073        "Return the number of elements in the sorted set ``name``"
1074        return self.execute_command('ZCARD', name)
1075
1076    def zcount(self, name, min, max):
1077        return self.execute_command('ZCOUNT', name, min, max)
1078
1079    def zincr(self, key, member, value=1):
1080        "This has been deprecated, use zincrby instead"
1081        warnings.warn(DeprecationWarning(
1082            "Redis.zincr has been deprecated, use Redis.zincrby instead"
1083            ))
1084        return self.zincrby(key, member, value)
1085
1086    def zincrby(self, name, value, amount=1):
1087        "Increment the score of ``value`` in sorted set ``name`` by ``amount``"
1088        return self.execute_command('ZINCRBY', name, amount, value)
1089
1090    def zinter(self, dest, keys, aggregate=None):
1091        warnings.warn(DeprecationWarning(
1092            "Redis.zinter has been deprecated, use Redis.zinterstore instead"
1093            ))
1094        return self.zinterstore(dest, keys, aggregate)
1095
1096    def zinterstore(self, dest, keys, aggregate=None):
1097        """
1098        Intersect multiple sorted sets specified by ``keys`` into
1099        a new sorted set, ``dest``. Scores in the destination will be
1100        aggregated based on the ``aggregate``, or SUM if none is provided.
1101        """
1102        return self._zaggregate('ZINTERSTORE', dest, keys, aggregate)
1103
1104    def zrange(self, name, start, end, desc=False, withscores=False):
1105        """
1106        Return a range of values from sorted set ``name`` between
1107        ``start`` and ``end`` sorted in ascending order.
1108
1109        ``start`` and ``end`` can be negative, indicating the end of the range.
1110
1111        ``desc`` indicates to sort in descending order.
1112
1113        ``withscores`` indicates to return the scores along with the values.
1114            The return type is a list of (value, score) pairs
1115        """
1116        if desc:
1117            return self.zrevrange(name, start, end, withscores)
1118        pieces = ['ZRANGE', name, start, end]
1119        if withscores:
1120            pieces.append('withscores')
1121        return self.execute_command(*pieces, **{'withscores': withscores})
1122
1123    def zrangebyscore(self, name, min, max,
1124            start=None, num=None, withscores=False):
1125        """
1126        Return a range of values from the sorted set ``name`` with scores
1127        between ``min`` and ``max``.
1128
1129        If ``start`` and ``num`` are specified, then return a slice of the range.
1130
1131        ``withscores`` indicates to return the scores along with the values.
1132            The return type is a list of (value, score) pairs
1133        """
1134        if (start is not None and num is None) or \
1135                (num is not None and start is None):
1136            raise RedisError("``start`` and ``num`` must both be specified")
1137        pieces = ['ZRANGEBYSCORE', name, min, max]
1138        if start is not None and num is not None:
1139            pieces.extend(['LIMIT', start, num])
1140        if withscores:
1141            pieces.append('withscores')
1142        return self.execute_command(*pieces, **{'withscores': withscores})
1143
1144    def zrank(self, name, value):
1145        """
1146        Returns a 0-based value indicating the rank of ``value`` in sorted set
1147        ``name``
1148        """
1149        return self.execute_command('ZRANK', name, value)
1150
1151    def zrem(self, name, value):
1152        "Remove member ``value`` from sorted set ``name``"
1153        return self.execute_command('ZREM', name, value)
1154
1155    def zremrangebyrank(self, name, min, max):
1156        """
1157        Remove all elements in the sorted set ``name`` with ranks between
1158        ``min`` and ``max``. Values are 0-based, ordered from smallest score
1159        to largest. Values can be negative indicating the highest scores.
1160        Returns the number of elements removed
1161        """
1162        return self.execute_command('ZREMRANGEBYRANK', name, min, max)
1163
1164    def zremrangebyscore(self, name, min, max):
1165        """
1166        Remove all elements in the sorted set ``name`` with scores
1167        between ``min`` and ``max``. Returns the number of elements removed.
1168        """
1169        return self.execute_command('ZREMRANGEBYSCORE', name, min, max)
1170
1171    def zrevrange(self, name, start, num, withscores=False):
1172        """
1173        Return a range of values from sorted set ``name`` between
1174        ``start`` and ``num`` sorted in descending order.
1175
1176        ``start`` and ``num`` can be negative, indicating the end of the range.
1177
1178        ``withscores`` indicates to return the scores along with the values
1179            as a dictionary of value => score
1180        """
1181        pieces = ['ZREVRANGE', name, start, num]
1182        if withscores:
1183            pieces.append('withscores')
1184        return self.execute_command(*pieces, **{'withscores': withscores})
1185
1186    def zrevrank(self, name, value):
1187        """
1188        Returns a 0-based value indicating the descending rank of
1189        ``value`` in sorted set ``name``
1190        """
1191        return self.execute_command('ZREVRANK', name, value)
1192
1193    def zscore(self, name, value):
1194        "Return the score of element ``value`` in sorted set ``name``"
1195        return self.execute_command('ZSCORE', name, value)
1196
1197    def zunion(self, dest, keys, aggregate=None):
1198        warnings.warn(DeprecationWarning(
1199            "Redis.zunion has been deprecated, use Redis.zunionstore instead"
1200            ))
1201        return self.zunionstore(dest, keys, aggregate)
1202
1203    def zunionstore(self, dest, keys, aggregate=None):
1204        """
1205        Union multiple sorted sets specified by ``keys`` into
1206        a new sorted set, ``dest``. Scores in the destination will be
1207        aggregated based on the ``aggregate``, or SUM if none is provided.
1208        """
1209        return self._zaggregate('ZUNIONSTORE', dest, keys, aggregate)
1210
1211    def _zaggregate(self, command, dest, keys, aggregate=None):
1212        pieces = [command, dest, len(keys)]
1213        if isinstance(keys, dict):
1214            items = keys.items()
1215            keys = [i[0] for i in items]
1216            weights = [i[1] for i in items]
1217        else:
1218            weights = None
1219        pieces.extend(keys)
1220        if weights:
1221            pieces.append('WEIGHTS')
1222            pieces.extend(weights)
1223        if aggregate:
1224            pieces.append('AGGREGATE')
1225            pieces.append(aggregate)
1226        return self.execute_command(*pieces)
1227
1228    #### HASH COMMANDS ####
1229    def hdel(self, name, key):
1230        "Delete ``key`` from hash ``name``"
1231        return self.execute_command('HDEL', name, key)
1232
1233    def hexists(self, name, key):
1234        "Returns a boolean indicating if ``key`` exists within hash ``name``"
1235        return self.execute_command('HEXISTS', name, key)
1236
1237    def hget(self, name, key):
1238        "Return the value of ``key`` within the hash ``name``"
1239        return self.execute_command('HGET', name, key)
1240
1241    def hgetall(self, name):
1242        "Return a Python dict of the hash's name/value pairs"
1243        return self.execute_command('HGETALL', name)
1244
1245    def hincrby(self, name, key, amount=1):
1246        "Increment the value of ``key`` in hash ``name`` by ``amount``"
1247        return self.execute_command('HINCRBY', name, key, amount)
1248
1249    def hkeys(self, name):
1250        "Return the list of keys within hash ``name``"
1251        return self.execute_command('HKEYS', name)
1252
1253    def hlen(self, name):
1254        "Return the number of elements in hash ``name``"
1255        return self.execute_command('HLEN', name)
1256
1257    def hset(self, name, key, value):
1258        """
1259        Set ``key`` to ``value`` within hash ``name``
1260        Returns 1 if HSET created a new field, otherwise 0
1261        """
1262        return self.execute_command('HSET', name, key, value)
1263
1264    def hsetnx(self, name, key, value):
1265        """
1266        Set ``key`` to ``value`` within hash ``name`` if ``key`` does not
1267        exist.  Returns 1 if HSETNX created a field, otherwise 0.
1268        """
1269        return self.execute_command("HSETNX", name, key, value)
1270
1271    def hmset(self, name, mapping):
1272        """
1273        Sets each key in the ``mapping`` dict to its corresponding value
1274        in the hash ``name``
1275        """
1276        items = []
1277        for pair in mapping.iteritems():
1278            items.extend(pair)
1279        return self.execute_command('HMSET', name, *items)
1280
1281    def hmget(self, name, keys):
1282        "Returns a list of values ordered identically to ``keys``"
1283        return self.execute_command('HMGET', name, *keys)
1284
1285    def hvals(self, name):
1286        "Return the list of values within hash ``name``"
1287        return self.execute_command('HVALS', name)
1288
1289
1290    # channels
1291    def psubscribe(self, patterns):
1292        "Subscribe to all channels matching any pattern in ``patterns``"
1293        if isinstance(patterns, basestring):
1294            patterns = [patterns]
1295        response = self.execute_command('PSUBSCRIBE', *patterns)
1296        # this is *after* the SUBSCRIBE in order to allow for lazy and broken
1297        # connections that need to issue AUTH and SELECT commands
1298        self.subscribed = True
1299        return response
1300
1301    def punsubscribe(self, patterns=[]):
1302        """
1303        Unsubscribe from any channel matching any pattern in ``patterns``.
1304        If empty, unsubscribe from all channels.
1305        """
1306        if isinstance(patterns, basestring):
1307            patterns = [patterns]
1308        return self.execute_command('PUNSUBSCRIBE', *patterns)
1309
1310    def subscribe(self, channels):
1311        "Subscribe to ``channels``, waiting for messages to be published"
1312        if isinstance(channels, basestring):
1313            channels = [channels]
1314        response = self.execute_command('SUBSCRIBE', *channels)
1315        # this is *after* the SUBSCRIBE in order to allow for lazy and broken
1316        # connections that need to issue AUTH and SELECT commands
1317        self.subscribed = True
1318        return response
1319
1320    def unsubscribe(self, channels=[]):
1321        """
1322        Unsubscribe from ``channels``. If empty, unsubscribe
1323        from all channels
1324        """
1325        if isinstance(channels, basestring):
1326            channels = [channels]
1327        return self.execute_command('UNSUBSCRIBE', *channels)
1328
1329    def publish(self, channel, message):
1330        """
1331        Publish ``message`` on ``channel``.
1332        Returns the number of subscribers the message was delivered to.
1333        """
1334        return self.execute_command('PUBLISH', channel, message)
1335
1336    def listen(self):
1337        "Listen for messages on channels this client has been subscribed to"
1338        while self.subscribed:
1339            r = self.parse_response('LISTEN')
1340            if r[0] == 'pmessage':
1341                msg = {
1342                'type': r[0],
1343                'pattern': r[1],
1344                'channel': r[2],
1345                'data': r[3]
1346                }
1347            else:
1348                msg = {
1349                'type': r[0],
1350                'pattern': None,
1351                'channel': r[1],
1352                'data': r[2]
1353                }
1354            if r[0] == 'unsubscribe' and r[2] == 0:
1355                self.subscribed = False
1356            yield msg
1357
1358    ### ALCHEMY DATABASE ###
1359    def createTable(self, tablename, column_defitions):
1360        return self.execute_command('CREATE', 'TABLE', tablename, "(" + column_defitions + ")")
1361    def dropTable(self, tablename):
1362        return self.execute_command('DROP', 'TABLE', tablename)
1363    def desc(self, tablename):
1364        return self.execute_command('DESC', tablename)
1365    def dump(self, tablename):
1366        return self.execute_command('DUMP', tablename)
1367    def dumpToMysql(self, tablename, mysqltablename):
1368        if mysqltablename == "":
1369            return self.execute_command('DUMP', tablename, "TO", "MYSQL")
1370        else:
1371            return self.execute_command('DUMP', tablename, "TO", "MYSQL", mysqltablename)
1372    def dumpToFile(self, tablename, filename):
1373        return self.execute_command('DUMP', tablename, "TO", "FILE", filename)
1374
1375    def createIndex(self, indexname, tablename, columnname):
1376        return self.execute_command('CREATE', 'INDEX', indexname, 'ON', tablename, "(" + columnname + ")")
1377    def dropIndex(self, indexname):
1378        return self.execute_command('DROP', 'INDEX', indexname)
1379
1380    def insert(self, tablename, values):
1381        return self.execute_command('INSERT', 'INTO', tablename, 'VALUES', "(" + values + ")")
1382    def insert_ret_size(self, tablename, values):
1383        return self.execute_command('INSERT', 'INTO', tablename, 'VALUES', "(" + values + ")", "RETURN", "SIZE")
1384    def sqlSelect(self, columns, tables, where_clause):
1385        return self.execute_command('SELECT', columns, 'FROM', tables, 'WHERE', where_clause)
1386    def scanSelect(self, columns, tables, where_clause):
1387        if where_clause == "":
1388            return self.execute_command('SCANSELECT', columns, 'FROM', tables)
1389        else:
1390            return self.execute_command('SCANSELECT', columns, 'FROM', tables, 'WHERE', where_clause)
1391    def update(self, tablename, value_list, where_clause):
1392        return self.execute_command('UPDATE', tablename, 'SET', value_list, 'WHERE', where_clause)
1393    def delete(self, tablename, where_clause):
1394        return self.execute_command('DELETE', 'FROM', tablename, 'WHERE', where_clause)
1395
1396    def createTableAs(self, tablename, statement):
1397        return self.execute_command('CREATE', 'TABLE', tablename, "AS " + statement)
1398    def lua(self, command):
1399        return self.execute_command('LUA', command)
1400    def norm(self, main_wildcard, secondary_wildcard):
1401        return self.execute_command('NORM', main_wildcard, secondary_wildcard)
1402    def norm(self, tablename, main_wildcard):
1403        return self.execute_command('DENORM', tablename, main_wildcard)
1404
1405
1406class Pipeline(Redis):
1407    """
1408    Pipelines provide a way to transmit multiple commands to the Redis server
1409    in one transmission.  This is convenient for batch processing, such as
1410    saving all the values in a list to Redis.
1411
1412    All commands executed within a pipeline are wrapped with MULTI and EXEC
1413    calls. This guarantees all commands executed in the pipeline will be
1414    executed atomically.
1415
1416    Any command raising an exception does *not* halt the execution of
1417    subsequent commands in the pipeline. Instead, the exception is caught
1418    and its instance is placed into the response list returned by execute().
1419    Code iterating over the response list should be able to deal with an
1420    instance of an exception as a potential value. In general, these will be
1421    ResponseError exceptions, such as those raised when issuing a command
1422    on a key of a different datatype.
1423    """
1424    def __init__(self, connection, transaction, charset, errors):
1425        self.connection = connection
1426        self.transaction = transaction
1427        self.encoding = charset
1428        self.errors = errors
1429        self.subscribed = False # NOTE not in use, but necessary
1430        self.reset()
1431
1432    def reset(self):
1433        self.command_stack = []
1434
1435    def _execute_command(self, command_name, command, **options):
1436        """
1437        Stage a command to be executed when execute() is next called
1438
1439        Returns the current Pipeline object back so commands can be
1440        chained together, such as:
1441
1442        pipe = pipe.set('foo', 'bar').incr('baz').decr('bang')
1443
1444        At some other point, you can then run: pipe.execute(),
1445        which will execute all commands queued in the pipe.
1446        """
1447        # if the command_name is 'AUTH' or 'SELECT', then this command
1448        # must have originated after a socket connection and a call to
1449        # _setup_connection(). run these commands immediately without
1450        # buffering them.
1451        ###if command_name in ('AUTH', 'SELECT'):
1452        ### LACHEMY DATABASE ###
1453        if command_name in ('AUTH'):
1454            return super(Pipeline, self)._execute_command(
1455                command_name, command, **options)
1456        else:
1457            self.command_stack.append((command_name, command, options))
1458        return self
1459
1460    def _execute_transaction(self, commands):
1461        # wrap the commands in MULTI ... EXEC statements to indicate an
1462        # atomic operation
1463        all_cmds = ''.join([c for _1, c, _2 in chain(
1464            (('', 'MULTI\r\n', ''),),
1465            commands,
1466            (('', 'EXEC\r\n', ''),)
1467            )])
1468        self.connection.send(all_cmds, self)
1469        # parse off the response for MULTI and all commands prior to EXEC
1470        for i in range(len(commands)+1):
1471            _ = self.parse_response('_')
1472        # parse the EXEC. we want errors returned as items in the response
1473        response = self.parse_response('_', catch_errors=True)
1474
1475        if response is None:
1476            raise WatchError("Watched variable changed.")
1477
1478        if len(response) != len(commands):
1479            raise ResponseError("Wrong number of response items from "
1480                "pipeline execution")
1481        # Run any callbacks for the commands run in the pipeline
1482        data = []
1483        for r, cmd in zip(response, commands):
1484            if not isinstance(r, Exception):
1485                if cmd[0] in self.RESPONSE_CALLBACKS:
1486                    r = self.RESPONSE_CALLBACKS[cmd[0]](r, **cmd[2])
1487            data.append(r)
1488        return data
1489
1490    def _execute_pipeline(self, commands):
1491        # build up all commands into a single request to increase network perf
1492        all_cmds = ''.join([c for _1, c, _2 in commands])
1493        self.connection.send(all_cmds, self)
1494        data = []
1495        for command_name, _, options in commands:
1496            data.append(
1497                self.parse_response(command_name, catch_errors=True, **options)
1498                )
1499        return data
1500
1501    def execute(self):
1502        "Execute all the commands in the current pipeline"
1503        stack = self.command_stack
1504        self.reset()
1505        if self.transaction:
1506            execute = self._execute_transaction
1507        else:
1508            execute = self._execute_pipeline
1509        try:
1510            return execute(stack)
1511        except ConnectionError:
1512            self.connection.disconnect()
1513            return execute(stack)
1514
1515    def select(self, *args, **kwargs):
1516        raise RedisError("Cannot select a different database from a pipeline")
1517
1518
1519class Lock(object):
1520    """
1521    A shared, distributed Lock. Using Redis for locking allows the Lock
1522    to be shared across processes and/or machines.
1523
1524    It's left to the user to resolve deadlock issues and make sure
1525    multiple clients play nicely together.
1526    """
1527
1528    LOCK_FOREVER = 2**31+1 # 1 past max unix time
1529
1530    def __init__(self, redis, name, timeout=None, sleep=0.1):
1531        """
1532        Create a new Lock instnace named ``name`` using the Redis client
1533        supplied by ``redis``.
1534
1535        ``timeout`` indicates a maximum life for the lock.
1536        By default, it will remain locked until release() is called.
1537
1538        ``sleep`` indicates the amount of time to sleep per loop iteration
1539        when the lock is in blocking mode and another client is currently
1540        holding the lock.
1541
1542        Note: If using ``timeout``, you should make sure all the hosts
1543        that are running clients are within the same timezone and are using
1544        a network time service like ntp.
1545        """
1546        self.redis = redis
1547        self.name = name
1548        self.acquired_until = None
1549        self.timeout = timeout
1550        self.sleep = sleep
1551
1552    def __enter__(self):
1553        return self.acquire()
1554
1555    def __exit__(self, exc_type, exc_value, traceback):
1556        self.release()
1557
1558    def acquire(self, blocking=True):
1559        """
1560        Use Redis to hold a shared, distributed lock named ``name``.
1561        Returns True once the lock is acquired.
1562
1563        If ``blocking`` is False, always return immediately. If the lock
1564        was acquired, return True, otherwise return False.
1565        """
1566        sleep = self.sleep
1567        timeout = self.timeout
1568        while 1:
1569            unixtime = int(time.time())
1570            if timeout:
1571                timeout_at = unixtime + timeout
1572            else:
1573                timeout_at = Lock.LOCK_FOREVER
1574            if self.redis.setnx(self.name, timeout_at):
1575                self.acquired_until = timeout_at
1576                return True
1577            # We want blocking, but didn't acquire the lock
1578            # check to see if the current lock is expired
1579            existing = long(self.redis.get(self.name) or 1)
1580            if existing < unixtime:
1581                # the previous lock is expired, attempt to overwrite it
1582                existing = long(self.redis.getset(self.name, timeout_at) or 1)
1583                if existing < unixtime:
1584                    # we successfully acquired the lock
1585                    self.acquired_until = timeout_at
1586                    return True
1587            if not blocking:
1588                return False
1589            time.sleep(sleep)
1590
1591    def release(self):
1592        "Releases the already acquired lock"
1593        if self.acquired_until is None:
1594            raise ValueError("Cannot release an unlocked lock")
1595        existing = long(self.redis.get(self.name) or 1)
1596        # if the lock time is in the future, delete the lock
1597        if existing >= self.acquired_until:
1598            self.redis.delete(self.name)
1599        self.acquired_until = None