Source code for platypush.plugins.redis
from redis import Redis
from platypush.plugins import Plugin, action
from platypush.utils import get_redis_conf
[docs]
class RedisPlugin(Plugin):
"""
Plugin to send messages on Redis queues.
See https://redis-py.readthedocs.io/en/latest/connections.html#redis.Redis
for supported parameters.
"""
[docs]
def __init__(self, *args, **kwargs):
super().__init__()
self.args = args
self.kwargs = kwargs or get_redis_conf()
def _get_redis(self):
return Redis(*self.args, **self.kwargs)
[docs]
@action
def send_message(self, queue: str, msg, *args, **kwargs):
"""
Send a message to a Redis queue.
:param queue: Queue name
:param msg: Message to be sent
:type msg: str, bytes, list, dict, :class:`platypush.message.Message`
:param args: Args passed to the Redis constructor (see
https://redis-py.readthedocs.io/en/latest/connections.html#redis.Redis)
:param kwargs: Kwargs passed to the Redis constructor (see
https://redis-py.readthedocs.io/en/latest/connections.html#redis.Redis)
"""
if args or kwargs:
redis = Redis(*args, **kwargs)
else:
redis = self._get_redis()
return redis.rpush(queue, str(msg))
[docs]
@action
def mget(self, keys, *args):
"""
:returns: The values specified in keys as a key/value dict (wraps MGET)
"""
return {
keys[i]: value.decode() if isinstance(value, bytes) else value
for (i, value) in enumerate(self._get_redis().mget(keys, *args))
}
[docs]
@action
def mset(self, **kwargs):
"""
Set key/values based on mapping (wraps MSET)
"""
try:
return self._get_redis().mset(**kwargs)
except TypeError:
# Commit https://github.com/andymccurdy/redis-py/commit/90a52dd5de111f0053bb3ebaa7c78f73a82a1e3e
# broke back-compatibility with the previous way of passing
# key-value pairs to mset directly on kwargs. This try-catch block
# is to support things on all the redis-py versions
return self._get_redis().mset(mapping=kwargs) # type: ignore
[docs]
@action
def expire(self, key, expiration):
"""
Set an expiration time in seconds for the specified key
:param key: Key to set to expire
:type key: str
:param expiration: Expiration timeout (in seconds)
:type expiration: int
"""
return self._get_redis().expire(key, expiration)
[docs]
@action
def delete(self, *args):
"""
Delete one or multiple keys
:param args: Keys to delete
"""
return self._get_redis().delete(*args)
# vim:sw=4:ts=4:et: