nameko-redis-utils
Redis dependency and utils for Nameko
Installation
Install
pip3 install nameko-redis-pyInstall latest version from Git:
pip install git+https://github.com/fraglab/nameko-redis-py.gitUsage
Redis client
from redis import StrictRedis
from nameko.rpc import rpc
from nameko.extensions import DependencyProvider
from nameko_redis import Redis, SharedRedis
class MyService:
name = "my_service"
redis = Redis() # type: StrictRedis
@rpc
def get_data(self, key):
return self.redis.get(key)
class MyDependencyProvider(DependencyProvider):
shared_redis = SharedRedis()
redis = None
def setup(self):
self.shared_redis.register_provider(self)
def start(self):
self.shared_redis.start()
self.redis = self.shared_redis.get_client()
def stop(self):
self.shared_redis.unregister_provider(self)
def my_logic(self):
data = self.redis.get("key")
# Some logic
return data
def get_dependency(self, worker_ctx):
return self
Utils
PubSub Response Listener
Problem: Waiting for key or value attribute in Redis without blocking connection
Solution: Use single pubsub connection for waiting keys
import json
from typing import Optional
from nameko.rpc import rpc
from nameko_redis.pubsub_response_listener import PubSubResponsesListener, IResponsesListener
# Must return tuple with response_key and deserialized object
def message_deserializer(raw_data) -> (str, Optional[object, dict]):
data = json.loads(raw_data)
return data['response_key'], data
class MyObj:
def __init__(self, name, state):
self.name = name
self.state = state
def event_validation(deserialized_obj: MyObj):
return deserialized_obj.state == 'done'
# WAIT NOT EXISTS KEY
class MyService:
listener = PubSubResponsesListener(message_deserializer) # type: IResponsesListener
@rpc
def get(self):
return self.listener.wait_for_response("RESPONSE_KEY", timeout=10, wait_not_exists_key=True)
# WAIT FOR CHANGE IN VALUE
class MyService2:
listener = PubSubResponsesListener(
message_deserializer, event_validation=event_validation) # type: IResponsesListener
@rpc
def get(self):
return self.listener.wait_for_response("RESPONSE_KEY", timeout=10)
Configuration
Nameko configuration example:
REDIS:
DEFAULT:
url: 'redis://localhost:6379/0'
options:
retry_on_timeout: True
decode_responses: True
Redis, SharedRedis arguments:
- url - specify configuration redis key, like "DEFAULT"
- **options - specify any StrictRedis optional arguments, like encoding, retry_on_timeout, etc.
from nameko_redis import Redis
class MyService:
name = "my_service"
redis = Redis() # type: StrictRedis
