diff options
| -rw-r--r-- | CHANGES | 13 | ||||
| -rw-r--r-- | README.md | 67 | ||||
| -rw-r--r-- | redis/client.py | 68 | ||||
| -rw-r--r-- | tests/pipeline.py | 43 | ||||
| -rw-r--r-- | tests/server_commands.py | 19 |
5 files changed, 181 insertions, 29 deletions
@@ -1,6 +1,17 @@ -* 2.6.3 (in development) +* 2.7.0 (in development) * Added BITOP and BITCOUNT commands. Thanks Mark Tozzi. * Added the TIME command. Thanks Jason Knight. + * Added support for LUA scripting. Thanks to Angus Peart, Drew Smathers, + Issac Kelly, Louis-Philippe Perron, Sean Bleier, Jeffrey Kaditz, and + Dvir Volk for various patches and contributions to this feature. + * Changed the default error handling in pipelines. By default, the first + error in a pipeline will now be raised. A new parameter to the + pipeline's execute, `raise_on_error`, can be set to False to keep the + old behavior of embeedding the exception instances in the result. + * Fixed a bug with pipelines where parse errors won't corrupt the + socket. + * Added the optional `number` argument to SRANDMEMBER for use with + Redis 2.6+ servers. * 2.6.2 * `from_url` is now available as a classmethod on client classes. Thanks Jon Parise for the patch. @@ -269,10 +269,71 @@ which is much easier to read: >>> r.transaction(client_side_incr, 'OUR-SEQUENCE-KEY') [True] -## Versioning scheme +## LUA Scripting + +redis-py supports the EVAL, EVALSHA, and SCRIPT commands. However, there are +a number of edge cases that make these commands tedious to use in real world +scenarios. Therefore, redis-py exposes a Script object that makes scripting +much easier to use. + +To create a Script instance, use the `register_script` function on a client +instance passing the LUA code as the first argument. `register_script` returns +a Script instance that you can use throughout your code. + +The following trivial LUA script accepts two parameters: the name of a key and a +multiplier value. The script fetches the value stored in the key, multiplies +it with the multiplier value and returns the result. + + >>> r = redis.StrictRedis() + >>> lua = """ + ... local value = redis.call('GET', KEYS[1]) + ... value = tonumber(value) + ... return value * ARGV[1]""" + >>> multiply = r.register_script(lua) + +`multiply` is now a Script instance that is invoked by calling it like a +function. Script instances accept the following optional arguments: + +* keys: A list of key names that the script will access. This becomes the + KEYS list in LUA. +* args: A list of argument values. This becomes the ARGV list in LUA. +* client: A redis-py Client or Pipeline instance that will invoke the + script. If client isn't specified, the client that intiially + created the Script instance (the one that `register_script` was + invoked from) will be used. + +Continuing the example from above: + + >>> r.set('foo', 2) + >>> multiply(keys=['foo'], args=[5]) + 10 + +The value of key 'foo' is set to 2. When multiply is invoked, the 'foo' key is +passed to the script along with the multiplier value of 5. LUA executes the +script and returns the result, 10. + +Script instances can be executed using a different client instance, even one +that points to a completely different Redis server. + + >>> r2 = redis.StrictRedis('redis2.example.com') + >>> r2.set('foo', 3) + >>> multiply(keys=['foo'], args=[5], client=r2) + 15 + +The Script object ensures that the LUA script is loaded into Redis's script +cache. In the event of a NOSCRIPT error, it will load the script and retry +executing it. + +Script objects can also be used in pipelines. The pipeline instance should be +passed as the client argument when calling the script. Care is taken to ensure +that the script is registered in Redis's script cache just prior to pipeline +execution. -redis-py is versioned after Redis. For example, redis-py 2.0.0 should -support all the commands available in Redis 2.0.0. + >>> pipe = r.pipeline() + >>> pipe.set('foo', 5) + >>> multiply(keys=['foo'], args=[5], client=pipe) + >>> pipe.execute() + [True, 25] Author ------ diff --git a/redis/client.py b/redis/client.py index 96fe53b..42a01dd 100644 --- a/redis/client.py +++ b/redis/client.py @@ -1,6 +1,7 @@ from __future__ import with_statement -from itertools import starmap +from itertools import chain, starmap import datetime +import sys import warnings import time as mod_time from redis._compat import (b, izip, imap, iteritems, dictkeys, dictvalues, @@ -940,9 +941,16 @@ class StrictRedis(object): "Remove and return a random member of set ``name``" return self.execute_command('SPOP', name) - def srandmember(self, name): - "Return a random member of set ``name``" - return self.execute_command('SRANDMEMBER', name) + def srandmember(self, name, number=None): + """ + If ``number`` is None, returns a random member of set ``name``. + + If ``number`` is supplied, returns a list of ``number`` random + memebers of set ``name``. Note this is only available when running + Redis 2.6+. + """ + args = number and [number] or [] + return self.execute_command('SRANDMEMBER', name, *args) def srem(self, name, *values): "Remove ``values`` from set ``name``" @@ -1642,27 +1650,40 @@ class BasePipeline(object): self.command_stack.append((args, options)) return self - def _execute_transaction(self, connection, commands): + def _execute_transaction(self, connection, commands, raise_on_error): + cmds = chain([(('MULTI', ), {})], commands, [(('EXEC', ), {})]) all_cmds = SYM_EMPTY.join( starmap(connection.pack_command, - [args for args, options in commands])) + [args for args, options in cmds])) connection.send_packed_command(all_cmds) - # we don't care about the multi/exec any longer - commands = commands[1:-1] - # parse off the response for MULTI and all commands prior to EXEC. - # the only data we care about is the response the EXEC - # which is the last command - for i in range(len(commands) + 1): - self.parse_response(connection, '_') + # parse off the response for MULTI + self.parse_response(connection, '_') + # and all the other commands + errors = [] + for i, _ in enumerate(commands): + try: + self.parse_response(connection, '_') + except ResponseError: + errors.append((i, sys.exc_info()[1])) + # parse the EXEC. response = self.parse_response(connection, '_') if response is None: raise WatchError("Watched variable changed.") + # put any parse errors into the response + for i, e in errors: + response.insert(i, e) + if len(response) != len(commands): raise ResponseError("Wrong number of response items from " "pipeline execution") + + # find any errors in the response and raise if necessary + if raise_on_error: + self.raise_first_error(response) + # We have to run response callbacks manually data = [] for r, cmd in izip(response, commands): @@ -1674,14 +1695,22 @@ class BasePipeline(object): data.append(r) return data - def _execute_pipeline(self, connection, commands): + def _execute_pipeline(self, connection, commands, raise_on_error): # build up all commands into a single request to increase network perf all_cmds = SYM_EMPTY.join( starmap(connection.pack_command, [args for args, options in commands])) connection.send_packed_command(all_cmds) - return [self.parse_response(connection, args[0], **options) - for args, options in commands] + response = [self.parse_response(connection, args[0], **options) + for args, options in commands] + if raise_on_error: + self.raise_first_error(response) + return response + + def raise_first_error(self, response): + for r in response: + if isinstance(r, ResponseError): + raise r def parse_response(self, connection, command_name, **options): result = StrictRedis.parse_response( @@ -1703,13 +1732,12 @@ class BasePipeline(object): if not exist: immediate('SCRIPT', 'LOAD', s.script, **{'parse': 'LOAD'}) - def execute(self): + def execute(self, raise_on_error=True): "Execute all the commands in the current pipeline" if self.scripts: self.load_scripts() stack = self.command_stack if self.transaction or self.explicit_transaction: - stack = [(('MULTI', ), {})] + stack + [(('EXEC', ), {})] execute = self._execute_transaction else: execute = self._execute_pipeline @@ -1723,7 +1751,7 @@ class BasePipeline(object): self.connection = conn try: - return execute(conn, stack) + return execute(conn, stack, raise_on_error) except ConnectionError: conn.disconnect() # if we were watching a variable, the watch is no longer valid @@ -1736,7 +1764,7 @@ class BasePipeline(object): "one or more keys") # otherwise, it's safe to retry since the transaction isn't # predicated on any state - return execute(conn, stack) + return execute(conn, stack, raise_on_error) finally: self.reset() diff --git a/tests/pipeline.py b/tests/pipeline.py index 89c5742..74d7f7b 100644 --- a/tests/pipeline.py +++ b/tests/pipeline.py @@ -61,12 +61,13 @@ class PipelineTestCase(unittest.TestCase): pipe.set('a', int(a) + 1) self.assertRaises(redis.WatchError, pipe.execute) - def test_invalid_command_in_pipeline(self): - # all commands but the invalid one should be excuted correctly + def test_exec_error_in_response(self): + # an invalid pipeline command at exec time adds the exception instance + # to the list of returned values self.client['c'] = 'a' with self.client.pipeline() as pipe: pipe.set('a', 1).set('b', 2).lpush('c', 3).set('d', 4) - result = pipe.execute() + result = pipe.execute(raise_on_error=False) self.assertEquals(result[0], True) self.assertEquals(self.client['a'], b('1')) @@ -83,6 +84,42 @@ class PipelineTestCase(unittest.TestCase): self.assertEquals(pipe.set('z', 'zzz').execute(), [True]) self.assertEquals(self.client['z'], b('zzz')) + def test_exec_error_raised(self): + self.client['c'] = 'a' + with self.client.pipeline() as pipe: + pipe.set('a', 1).set('b', 2).lpush('c', 3).set('d', 4) + self.assertRaises(redis.ResponseError, pipe.execute) + + # make sure the pipe was restored to a working state + self.assertEquals(pipe.set('z', 'zzz').execute(), [True]) + self.assertEquals(self.client['z'], b('zzz')) + + def test_parse_error_in_response(self): + with self.client.pipeline() as pipe: + # the zrem is invalid because we don't pass any keys to it + pipe.set('a', 1).zrem('b').set('b', 2) + result = pipe.execute(raise_on_error=False) + + self.assertEquals(result[0], True) + self.assertEquals(self.client['a'], b('1')) + self.assert_(isinstance(result[1], redis.ResponseError)) + self.assertEquals(result[2], True) + self.assertEquals(self.client['b'], b('2')) + + # make sure the pipe was restored to a working state + self.assertEquals(pipe.set('z', 'zzz').execute(), [True]) + self.assertEquals(self.client['z'], b('zzz')) + + def test_parse_error_raised(self): + with self.client.pipeline() as pipe: + # the zrem is invalid because we don't pass any keys to it + pipe.set('a', 1).zrem('b').set('b', 2) + self.assertRaises(redis.ResponseError, pipe.execute) + + # make sure the pipe was restored to a working state + self.assertEquals(pipe.set('z', 'zzz').execute(), [True]) + self.assertEquals(self.client['z'], b('zzz')) + def test_watch_succeed(self): self.client.set('a', 1) self.client.set('b', 2) diff --git a/tests/server_commands.py b/tests/server_commands.py index b7dcf53..e8f6c7a 100644 --- a/tests/server_commands.py +++ b/tests/server_commands.py @@ -550,7 +550,11 @@ class ServerCommandsTestCase(unittest.TestCase): del self.client['a'] # real logic version = self.client.info()['redis_version'] - if StrictVersion(version) >= StrictVersion('1.3.4'): + if StrictVersion(version) >= StrictVersion('2.4.0'): + self.assertEqual(1, self.client.lpush('a', 'b')) + self.assertEqual(2, self.client.lpush('a', 'a')) + self.assertEqual(4, self.client.lpush('a', 'b', 'a')) + elif StrictVersion(version) >= StrictVersion('1.3.4'): self.assertEqual(1, self.client.lpush('a', 'b')) self.assertEqual(2, self.client.lpush('a', 'a')) else: @@ -684,7 +688,11 @@ class ServerCommandsTestCase(unittest.TestCase): del self.client['a'] # real logic version = self.client.info()['redis_version'] - if StrictVersion(version) >= StrictVersion('1.3.4'): + if StrictVersion(version) >= StrictVersion('2.4.0'): + self.assertEqual(1, self.client.rpush('a', 'a')) + self.assertEqual(2, self.client.rpush('a', 'b')) + self.assertEqual(4, self.client.rpush('a', 'a', 'b')) + elif StrictVersion(version) >= StrictVersion('1.3.4'): self.assertEqual(1, self.client.rpush('a', 'a')) self.assertEqual(2, self.client.rpush('a', 'b')) else: @@ -861,6 +869,13 @@ class ServerCommandsTestCase(unittest.TestCase): self.make_set('a', 'abc') self.assert_(self.client.srandmember('a') in b('abc')) + version = self.client.info()['redis_version'] + if StrictVersion(version) >= StrictVersion('2.6.0'): + randoms = self.client.srandmember('a', number=2) + self.assertEquals(len(randoms), 2) + for r in randoms: + self.assert_(r in b('abc')) + def test_srem(self): # key is not set self.assertEquals(self.client.srem('a', 'a'), False) |