Source code for catcher_modules.cache.redis

from catcher.steps.external_step import ExternalStep
from catcher.steps.step import update_variables

[docs]class Redis(ExternalStep): """ Work with Redis cache. Put value to cache or get it, increment/decrement or delete. :Input: :conf: redis configuration. Is an object. - host: redis host. Default is localhost - port: redis port. Default is 6379 - db: redis database number. Default is 0 :<command>: - command to run. Every command can have a list of arguments. Refer to `Redis <>`_ and `Redis-Py <>`_ :Examples: Set value (default configuration) :: variables: complex: a: 1 b: 'c' d: [1,2,4] redis: request: set: key: '{{ complex }}' Get value by key 'key' and register in variable 'var' :: redis: request: get: 'key' register: {var: '{{ OUTPUT }}'} Decrement, increment by 5 and delete :: redis: actions: - request: set: 'foo': 11 - request: decr: foo - request: incrby: foo: 5 - request: delete: - foo """ @update_variables def action(self, includes: dict, variables: dict) -> any: import redis body = self.simple_input(variables) in_data = body['request'] conf = in_data.get('conf', {}) r = redis.StrictRedis(host=conf.get('host', 'localhost'), port=conf.get('port', 6379), db=conf.get('db', 0), max_connections=1) [command] = [k for k in in_data.keys() if k != 'conf'] value = in_data.get(command, []) if isinstance(value, dict): # set: {key: value} flatlist = [str(item) for k in value for item in (k, value[k])] # convert to str to avoid data errors result = getattr(r, command.lower())(*flatlist) elif isinstance(value, str): # get: key result = getattr(r, command.lower())(value) else: # ??? result = getattr(r, command.lower())(*value) if isinstance(result, bytes): return variables, result.decode() return variables, result