Source code for redis_admin.client

import redis
import logging
from redis import sentinel

from . import settings


logger = logging.getLogger(__name__)
_sentinel = None
masters = dict()
slaves = dict()


[docs]def get_sentinel(): global _sentinel if not _sentinel: options = settings.SENTINEL_OPTIONS options.setdefault('socket_timeout', settings.SOCKET_TIMEOUT) _sentinel = sentinel.Sentinel(settings.SENTINELS, **options) return _sentinel
[docs]def get_master(name) -> redis.Redis: server = settings.SERVERS.get(name, {}).copy() if 'service_name' in server: server.setdefault('socket_timeout', settings.SOCKET_TIMEOUT) logger.debug('Getting master from sentinel %s: %r', name, server) server.pop('meta', dict()) return get_sentinel().master_for(**server) else: if name not in masters: if 'master' in server: server = server['master'] server.pop('meta', dict()) server.setdefault('socket_timeout', settings.SOCKET_TIMEOUT) logger.debug('Connecting to master %s: %r', name, server) masters[name] = redis.Redis(**server) return masters[name]
[docs]def get_slave(name) -> redis.Redis: server = settings.SERVERS.get(name, {}) if 'service_name' in server: server.setdefault('socket_timeout', settings.SOCKET_TIMEOUT) logger.debug('Getting slave from sentinel %s: %r', name, server) server.pop('meta', dict()) return get_sentinel().slave_for(**server) else: if name not in slaves: if 'slave' in server: server = server['slave'] server.pop('meta', dict()) server.setdefault('socket_timeout', settings.SOCKET_TIMEOUT) logger.debug('Connecting to slave %s: %r', name, server) slaves[name] = redis.Redis(**server) return slaves[name]