Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Support redis-py v2 and v3 #948

Merged
merged 1 commit into from
Nov 19, 2018
Merged

Support redis-py v2 and v3 #948

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion kombu/transport/redis.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,8 +146,15 @@ def __init__(self, *args, **kwargs):
def append(self, message, delivery_tag):
delivery = message.delivery_info
EX, RK = delivery['exchange'], delivery['routing_key']
# TODO: Remove this once we soley on Redis-py 3.0.0+
if redis.VERSION[0] >= 3:
thedrow marked this conversation as resolved.
Show resolved Hide resolved
# Redis-py changed the format of zadd args in v3.0.0
zadd_args = [{time(): delivery_tag}]
Copy link
Contributor

@Tenzer Tenzer Nov 20, 2018 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this not the wrong way round? The documentation says:

For ZADD, the dict is a mapping of element-names -> score.

So the key in the dict should be the element-name - in this case delivery_tag - and the value should be the score, which here is time()?

I did a quick test locally to check:

>>> from redis import Redis
>>> r = Redis()
>>> from time import time
>>> zadd_args = [{time(): 'test_tag'}]
>>> r.zadd('kombu', *zadd_args)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/jeppe/.virtualenvs/tmp-82c2c5e4c227252/lib/python3.6/site-packages/redis/client.py", line 2266, in zadd
    return self.execute_command('ZADD', name, *pieces, **options)
  File "/Users/jeppe/.virtualenvs/tmp-82c2c5e4c227252/lib/python3.6/site-packages/redis/client.py", line 755, in execute_command
    return self.parse_response(connection, command_name, **options)
  File "/Users/jeppe/.virtualenvs/tmp-82c2c5e4c227252/lib/python3.6/site-packages/redis/client.py", line 768, in parse_response
    response = connection.read_response()
  File "/Users/jeppe/.virtualenvs/tmp-82c2c5e4c227252/lib/python3.6/site-packages/redis/connection.py", line 638, in read_response
    raise response
redis.exceptions.ResponseError: value is not a valid float
>>> zadd_args2 = [{'test_tag': time()}]
>>> r.zadd('kombu', *zadd_args2)
1

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't know - even on redis 2.10.6 I get this:

>>> import redis
>>> redis.VERSION
(2, 10, 6)
>>> from time import time
>>> r = redis.Redis()
>>> r.zadd('kombu', time(), 'test_tag')
... 
redis.exceptions.ResponseError: value is not a valid float

and this is definately the same order as the code was previously.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, that's due to the difference between the Redis client and StrictRedis. In redis-py 3 StrictRedis has been renamed to Redis and StrictRedis has become an alias to Redis.

In redis-py 2 there's a difference in the input order, it's mentioned slightly in the README for the 2.10.6 release: https://github.com/andymccurdy/redis-py/tree/2.10.6#api-reference, but it might be easier to compare the code used. This is the StrictRedis.zadd method: https://github.com/andymccurdy/redis-py/blob/2.10.6/redis/client.py#L1677-L1697 and here's the Redis.zadd method: https://github.com/andymccurdy/redis-py/blob/2.10.6/redis/client.py#L2292-L2321.

>>> import redis
>>> redis.VERSION
(2, 10, 6)
>>> r = redis.StrictRedis()
>>> from time import time
>>> r.zadd('kombu', time(), 'test_tag')
1

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The order was flipped around in #953.

else:
zadd_args = [time(), delivery_tag]

with self.pipe_or_acquire() as pipe:
pipe.zadd(self.unacked_index_key, time(), delivery_tag) \
pipe.zadd(self.unacked_index_key, *zadd_args) \
.hset(self.unacked_key, delivery_tag,
dumps([message._raw, EX, RK])) \
.execute()
Expand Down
27 changes: 24 additions & 3 deletions t/unit/transport/test_redis.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,8 +75,15 @@ def hdel(self, key, k):
def sadd(self, key, member, *args):
self.sets[key].add(member)

def zadd(self, key, score1, member1, *args):
self.sets[key].add(member1)
def zadd(self, key, *args):
if redis.redis.VERSION[0] >= 3:
(mapping,) = args
for item in mapping:
self.sets[key].add(item)
else:
# TODO: remove me when we drop support for Redis-py v2
(score1, member1) = args
self.sets[key].add(member1)

def smembers(self, key):
return self.sets.get(key, set())
Expand Down Expand Up @@ -840,7 +847,21 @@ def setup(self):
def teardown(self):
self.connection.close()

def test_publish__get(self):
@mock.replace_module_value(redis.redis, 'VERSION', [3, 0, 0])
def test_publish__get_redispyv3(self):
channel = self.connection.channel()
producer = Producer(channel, self.exchange, routing_key='test_Redis')
self.queue(channel).declare()

producer.publish({'hello': 'world'})

assert self.queue(channel).get().payload == {'hello': 'world'}
assert self.queue(channel).get() is None
assert self.queue(channel).get() is None
assert self.queue(channel).get() is None

@mock.replace_module_value(redis.redis, 'VERSION', [2, 5, 10])
def test_publish__get_redispyv2(self):
channel = self.connection.channel()
producer = Producer(channel, self.exchange, routing_key='test_Redis')
self.queue(channel).declare()
Expand Down