From 7abaabfb32460f2bc46d74858c16997864fc2cea Mon Sep 17 00:00:00 2001
From: Dong Zhang <41927498+dzhangalibaba@users.noreply.github.com>
Date: Thu, 10 Oct 2019 16:26:02 -0700
Subject: [PATCH] [multi-DB] Part 3: Python API changes (#52)
* multiDB part3 : python API
* add validation check and update some minor suggestions
---
src/swsssdk/__init__.py | 26 +---
src/swsssdk/config/database.json | 31 ----
src/swsssdk/config/database_config.json | 57 ++++++++
src/swsssdk/configdb.py | 26 ++--
src/swsssdk/dbconnector.py | 184 +++++++++++++++++++++++-
src/swsssdk/interface.py | 175 ++++++++++------------
test/test_moduleLoad.py | 10 +-
7 files changed, 328 insertions(+), 181 deletions(-)
delete mode 100644 src/swsssdk/config/database.json
create mode 100644 src/swsssdk/config/database_config.json
diff --git a/src/swsssdk/__init__.py b/src/swsssdk/__init__.py
index 1989c93937cc..1722bb70182b 100644
--- a/src/swsssdk/__init__.py
+++ b/src/swsssdk/__init__.py
@@ -7,32 +7,8 @@
logger.setLevel(logging.INFO)
logger.addHandler(logging.NullHandler())
-import json
-import os
-
-
-def _load_connector_map():
- """
- Get database map from the packaged config.
- """
- global _connector_map
- db_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'config', 'database.json')
-
- try:
- with open(db_file, 'r') as f:
- _connector_map = json.load(f)
- except (OSError, IOError):
- # Missing configuration means we can't configure the database index. Fatal error.
- msg = "Could not open database index '{}'".format(db_file)
- logger.exception(msg)
- raise RuntimeError(msg)
-
-
-_connector_map = {}
-_load_connector_map()
-
try:
- from .dbconnector import SonicV2Connector
+ from .dbconnector import SonicDBConfig, SonicV2Connector
from .configdb import ConfigDBConnector
except (KeyError, ValueError):
msg = "Failed to database connector objects -- incorrect database config schema."
diff --git a/src/swsssdk/config/database.json b/src/swsssdk/config/database.json
deleted file mode 100644
index 640012f15945..000000000000
--- a/src/swsssdk/config/database.json
+++ /dev/null
@@ -1,31 +0,0 @@
-{
- "SonicV2Connector": {
- "db_map": {
- "APPL_DB": {
- "db": 0,
- "separator": ":"
- },
- "ASIC_DB": {
- "db": 1,
- "separator": ":"
- },
- "COUNTERS_DB": {
- "db": 2,
- "separator": ":"
- },
- "CONFIG_DB": {
- "db": 4,
- "separator": "|"
- },
- "STATE_DB": {
- "db": 6,
- "separator": "|"
- },
- "SNMP_OVERLAY_DB": {
- "db": 7,
- "separator": "|"
-
- }
- }
- }
-}
diff --git a/src/swsssdk/config/database_config.json b/src/swsssdk/config/database_config.json
new file mode 100644
index 000000000000..b86ae11bba98
--- /dev/null
+++ b/src/swsssdk/config/database_config.json
@@ -0,0 +1,57 @@
+{
+ "INSTANCES": {
+ "redis":{
+ "hostname" : "127.0.0.1",
+ "port" : 6379,
+ "unix_socket_path" : "/var/run/redis/redis.sock"
+ }
+ },
+ "DATABASES" : {
+ "APPL_DB" : {
+ "id" : 0,
+ "separator": ":",
+ "instance" : "redis"
+ },
+ "ASIC_DB" : {
+ "id" : 1,
+ "separator": ":",
+ "instance" : "redis"
+ },
+ "COUNTERS_DB" : {
+ "id" : 2,
+ "separator": ":",
+ "instance" : "redis"
+ },
+ "LOGLEVEL_DB" : {
+ "id" : 3,
+ "separator": ":",
+ "instance" : "redis"
+ },
+ "CONFIG_DB" : {
+ "id" : 4,
+ "separator": "|",
+ "instance" : "redis"
+ },
+ "PFC_WD_DB" : {
+ "id" : 5,
+ "separator": ":",
+ "instance" : "redis"
+ },
+ "FLEX_COUNTER_DB" : {
+ "id" : 5,
+ "separator": ":",
+ "instance" : "redis"
+ },
+ "STATE_DB" : {
+ "id" : 6,
+ "separator": "|",
+ "instance" : "redis"
+ },
+ "SNMP_OVERLAY_DB" : {
+ "id" : 7,
+ "separator": "|",
+ "instance" : "redis"
+ }
+ },
+ "VERSION" : "1.0"
+}
diff --git a/src/swsssdk/configdb.py b/src/swsssdk/configdb.py
index 2fa896f25ab5..fe77c5838a8a 100644
--- a/src/swsssdk/configdb.py
+++ b/src/swsssdk/configdb.py
@@ -37,11 +37,11 @@ def __init__(self, **kwargs):
self.handlers = {}
def __wait_for_db_init(self):
- client = self.redis_clients[self.db_name]
+ client = self.get_redis_client(self.db_name)
pubsub = client.pubsub()
initialized = client.get(self.INIT_INDICATOR)
if not initialized:
- pattern = "__keyspace@{}__:{}".format(self.db_map[self.db_name]['db'], self.INIT_INDICATOR)
+ pattern = "__keyspace@{}__:{}".format(self.get_dbid(self.db_name), self.INIT_INDICATOR)
pubsub.psubscribe(pattern)
for item in pubsub.listen():
if item['type'] == 'pmessage':
@@ -55,7 +55,7 @@ def __wait_for_db_init(self):
def db_connect(self, dbname, wait_for_init=False, retry_on=False):
self.db_name = dbname
- self.KEY_SEPARATOR = self.TABLE_NAME_SEPARATOR = self.db_map[self.db_name]['separator']
+ self.KEY_SEPARATOR = self.TABLE_NAME_SEPARATOR = self.get_db_separator(self.db_name)
SonicV2Connector.connect(self, self.db_name, retry_on)
if wait_for_init:
self.__wait_for_db_init()
@@ -89,15 +89,15 @@ def __fire(self, table, key, data):
def listen(self):
"""Start listen Redis keyspace events and will trigger corresponding handlers when content of a table changes.
"""
- self.pubsub = self.redis_clients[self.db_name].pubsub()
- self.pubsub.psubscribe("__keyspace@{}__:*".format(self.db_map[self.db_name]['db']))
+ self.pubsub = self.get_redis_client(self.db_name).pubsub()
+ self.pubsub.psubscribe("__keyspace@{}__:*".format(self.get_dbid(self.db_name)))
for item in self.pubsub.listen():
if item['type'] == 'pmessage':
key = item['channel'].split(':', 1)[1]
try:
(table, row) = key.split(self.TABLE_NAME_SEPARATOR, 1)
if self.handlers.has_key(table):
- client = self.redis_clients[self.db_name]
+ client = self.get_redis_client(self.db_name)
data = self.__raw_to_typed(client.hgetall(key))
self.__fire(table, row, data)
except ValueError:
@@ -171,7 +171,7 @@ def set_entry(self, table, key, data):
Pass None as data will delete the entry.
"""
key = self.serialize_key(key)
- client = self.redis_clients[self.db_name]
+ client = self.get_redis_client(self.db_name)
_hash = '{}{}{}'.format(table.upper(), self.TABLE_NAME_SEPARATOR, key)
if data == None:
client.delete(_hash)
@@ -193,7 +193,7 @@ def mod_entry(self, table, key, data):
Pass None as data will delete the entry.
"""
key = self.serialize_key(key)
- client = self.redis_clients[self.db_name]
+ client = self.get_redis_client(self.db_name)
_hash = '{}{}{}'.format(table.upper(), self.TABLE_NAME_SEPARATOR, key)
if data == None:
client.delete(_hash)
@@ -210,7 +210,7 @@ def get_entry(self, table, key):
Empty dictionary if table does not exist or entry does not exist.
"""
key = self.serialize_key(key)
- client = self.redis_clients[self.db_name]
+ client = self.get_redis_client(self.db_name)
_hash = '{}{}{}'.format(table.upper(), self.TABLE_NAME_SEPARATOR, key)
return self.__raw_to_typed(client.hgetall(_hash))
@@ -223,7 +223,7 @@ def get_keys(self, table, split=True):
Returns:
List of keys.
"""
- client = self.redis_clients[self.db_name]
+ client = self.get_redis_client(self.db_name)
pattern = '{}{}*'.format(table.upper(), self.TABLE_NAME_SEPARATOR)
keys = client.keys(pattern)
data = []
@@ -250,7 +250,7 @@ def get_table(self, table):
or { ('l1_key', 'l2_key', ...): {'column_key': value, ...}, ...} for a multi-key table.
Empty dictionary if table does not exist.
"""
- client = self.redis_clients[self.db_name]
+ client = self.get_redis_client(self.db_name)
pattern = '{}{}*'.format(table.upper(), self.TABLE_NAME_SEPARATOR)
keys = client.keys(pattern)
data = {}
@@ -274,7 +274,7 @@ def delete_table(self, table):
Args:
table: Table name.
"""
- client = self.redis_clients[self.db_name]
+ client = self.get_redis_client(self.db_name)
pattern = '{}{}*'.format(table.upper(), self.TABLE_NAME_SEPARATOR)
keys = client.keys(pattern)
data = {}
@@ -307,7 +307,7 @@ def get_config(self):
...
}
"""
- client = self.redis_clients[self.db_name]
+ client = self.get_redis_client(self.db_name)
keys = client.keys('*')
data = {}
for key in keys:
diff --git a/src/swsssdk/dbconnector.py b/src/swsssdk/dbconnector.py
index ec830028e97c..32c895d37a64 100644
--- a/src/swsssdk/dbconnector.py
+++ b/src/swsssdk/dbconnector.py
@@ -1,20 +1,190 @@
"""
Database connection module for SwSS
"""
-from . import _connector_map, logger
+from . import logger
from .interface import DBInterface
-
+import os
+import json
# FIXME: Convert to metaclasses when Py2 support is removed. Metaclasses have unique interfaces to Python2/Python3.
+class SonicDBConfig(object):
+ SONIC_DB_CONFIG_FILE = "/var/run/redis/sonic-db/database_config.json"
+ _sonic_db_config_init = False
+ _sonic_db_config = {}
+
+ @staticmethod
+ def load_sonic_db_config(sonic_db_file_path=SONIC_DB_CONFIG_FILE):
+ """
+ Get multiple database config from the database_config.json
+ """
+ if SonicDBConfig._sonic_db_config_init == True:
+ return
+
+ try:
+ if os.path.isfile(sonic_db_file_path) == False:
+ msg = "'{}' is not found, it is not expected in production devices!!".format(sonic_db_file_path)
+ logger.warning(msg)
+ sonic_db_file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'config', 'database_config.json')
+ with open(sonic_db_file_path, "r") as read_file:
+ SonicDBConfig._sonic_db_config = json.load(read_file)
+ except (OSError, IOError):
+ msg = "Could not open sonic database config file '{}'".format(sonic_db_file_path)
+ logger.exception(msg)
+ raise RuntimeError(msg)
+ SonicDBConfig._sonic_db_config_init = True
+
+ @staticmethod
+ def db_name_validation(db_name):
+ if SonicDBConfig._sonic_db_config_init == False:
+ SonicDBConfig.load_sonic_db_config()
+ if db_name not in SonicDBConfig._sonic_db_config["DATABASES"]:
+ msg = "{} is not a valid database name in configuration file".format(db_name)
+ logger.exception(msg)
+ raise RuntimeError(msg)
+
+ @staticmethod
+ def inst_name_validation(inst_name):
+ if SonicDBConfig._sonic_db_config_init == False:
+ SonicDBConfig.load_sonic_db_config()
+ if inst_name not in SonicDBConfig._sonic_db_config["INSTANCES"]:
+ msg = "{} is not a valid instance name in configuration file".format(inst_name)
+ logger.exception(msg)
+ raise RuntimeError(msg)
+
+ @staticmethod
+ def get_dblist():
+ if SonicDBConfig._sonic_db_config_init == False:
+ SonicDBConfig.load_sonic_db_config()
+ return SonicDBConfig._sonic_db_config["DATABASES"].keys()
+
+ @staticmethod
+ def get_instance(db_name):
+ if SonicDBConfig._sonic_db_config_init == False:
+ SonicDBConfig.load_sonic_db_config()
+ SonicDBConfig.db_name_validation(db_name)
+ inst_name = SonicDBConfig._sonic_db_config["DATABASES"][db_name]["instance"]
+ SonicDBConfig.inst_name_validation(inst_name)
+ return SonicDBConfig._sonic_db_config["INSTANCES"][inst_name]
+
+ @staticmethod
+ def get_socket(db_name):
+ if SonicDBConfig._sonic_db_config_init == False:
+ SonicDBConfig.load_sonic_db_config()
+ SonicDBConfig.db_name_validation(db_name)
+ return SonicDBConfig.get_instance(db_name)["unix_socket_path"]
+
+ @staticmethod
+ def get_hostname(db_name):
+ if SonicDBConfig._sonic_db_config_init == False:
+ SonicDBConfig.load_sonic_db_config()
+ SonicDBConfig.db_name_validation(db_name)
+ return SonicDBConfig.get_instance(db_name)["hostname"]
+
+ @staticmethod
+ def get_port(db_name):
+ if SonicDBConfig._sonic_db_config_init == False:
+ SonicDBConfig.load_sonic_db_config()
+ SonicDBConfig.db_name_validation(db_name)
+ return SonicDBConfig.get_instance(db_name)["port"]
+
+ @staticmethod
+ def get_dbid(db_name):
+ if SonicDBConfig._sonic_db_config_init == False:
+ SonicDBConfig.load_sonic_db_config()
+ SonicDBConfig.db_name_validation(db_name)
+ return SonicDBConfig._sonic_db_config["DATABASES"][db_name]["id"]
+
+ @staticmethod
+ def get_separator(db_name):
+ if SonicDBConfig._sonic_db_config_init == False:
+ SonicDBConfig.load_sonic_db_config()
+ SonicDBConfig.db_name_validation(db_name)
+ return SonicDBConfig._sonic_db_config["DATABASES"][db_name]["separator"]
+
class SonicV2Connector(DBInterface):
- def __init__(self, **kwargs):
+ def __init__(self, use_unix_socket_path=True, **kwargs):
super(SonicV2Connector, self).__init__(**kwargs)
+ self.use_unix_socket_path = use_unix_socket_path
+ for db_name in self.get_db_list():
+ # set a database name as a constant value attribute.
+ setattr(self, db_name, db_name)
- pass
+ def connect(self, db_name, retry_on=True):
+ if self.use_unix_socket_path == False:
+ self.redis_kwargs["unix_socket_path"] = self.get_db_socket(db_name)
+ self.redis_kwargs["host"] = None
+ self.redis_kwargs["port"] = None
+ else:
+ self.redis_kwargs["host"] = self.get_db_hostname(db_name)
+ self.redis_kwargs["port"] = self.get_db_port(db_name)
+ self.redis_kwargs["unix_socket_path"] = None
+ db_id = self.get_dbid(db_name)
+ super(SonicV2Connector, self).connect(db_id, retry_on)
+
+ def close(self, db_name):
+ db_id = self.get_dbid(db_name)
+ super(SonicV2Connector, self).close(db_id)
+
+ def get_db_list(self):
+ return SonicDBConfig.get_dblist()
+ def get_db_instance(self, db_name):
+ return SonicDBConfig.get_instance(db_name)
-SonicV2Connector.db_map = _connector_map[SonicV2Connector.__name__]['db_map']
+ def get_db_socket(self, db_name):
+ return SonicDBConfig.get_socket(db_name)
-if len(SonicV2Connector.db_map) != len({v['db'] for k, v in SonicV2Connector.db_map.items()}):
- raise RuntimeError("Duplicate DB index detected in configuration.")
+ def get_db_hostname(self, db_name):
+ return SonicDBConfig.get_hostname(db_name)
+
+ def get_db_port(self, db_name):
+ return SonicDBConfig.get_port(db_name)
+
+ def get_dbid(self, db_name):
+ return SonicDBConfig.get_dbid(db_name)
+
+ def get_db_separator(self, db_name):
+ return SonicDBConfig.get_separator(db_name)
+
+ def get_redis_client(self, db_name):
+ db_id = self.get_dbid(db_name)
+ return super(SonicV2Connector, self).get_redis_client(db_id)
+
+ def publish(self, db_name, channel, message):
+ db_id = self.get_dbid(db_name)
+ return super(SonicV2Connector, self).publish(db_id, channel, message)
+
+ def expire(self, db_name, key, timeout_sec):
+ db_id = self.get_dbid(db_name)
+ return super(SonicV2Connector, self).expire(db_id, key, timeout_sec)
+
+ def exists(self, db_name, key):
+ db_id = self.get_dbid(db_name)
+ return super(SonicV2Connector, self).exists(db_id, key)
+
+ def keys(self, db_name, pattern='*', *args, **kwargs):
+ db_id = self.get_dbid(db_name)
+ return super(SonicV2Connector, self).keys(db_id, pattern, *args, **kwargs)
+
+ def get(self, db_name, _hash, key, *args, **kwargs):
+ db_id = self.get_dbid(db_name)
+ return super(SonicV2Connector, self).get(db_id, _hash, key, *args, **kwargs)
+
+ def get_all(self, db_name, _hash, *args, **kwargs):
+ db_id = self.get_dbid(db_name)
+ return super(SonicV2Connector, self).get_all(db_id, _hash, *args, **kwargs)
+
+ def set(self, db_name, _hash, key, val, *args, **kwargs):
+ db_id = self.get_dbid(db_name)
+ return super(SonicV2Connector, self).set(db_id, _hash, key, val, *args, **kwargs)
+
+ def delete(self, db_name, key, *args, **kwargs):
+ db_id = self.get_dbid(db_name)
+ return super(SonicV2Connector, self).delete(db_id, key, *args, **kwargs)
+
+ def delete_all_by_pattern(self, db_name, pattern, *args, **kwargs):
+ db_id = self.get_dbid(db_name)
+ super(SonicV2Connector, self).delete_all_by_pattern(db_id, pattern, *args, **kwargs)
+
+ pass
diff --git a/src/swsssdk/interface.py b/src/swsssdk/interface.py
index c77530b719ed..a3a13cb1b17a 100644
--- a/src/swsssdk/interface.py
+++ b/src/swsssdk/interface.py
@@ -18,7 +18,7 @@ def blockable(f):
class SonicV2Connector:
@blockable
- def keys(self, db_name):
+ def keys(self, db_id):
# ...
# call with:
@@ -29,26 +29,26 @@ def keys(self, db_name):
"""
@wraps(f)
- def wrapped(inst, db_name, *args, **kwargs):
+ def wrapped(inst, db_id, *args, **kwargs):
blocking = kwargs.pop('blocking', False)
attempts = 0
while True:
try:
- ret_data = f(inst, db_name, *args, **kwargs)
- inst._unsubscribe_keyspace_notification(db_name)
+ ret_data = f(inst, db_id, *args, **kwargs)
+ inst._unsubscribe_keyspace_notification(db_id)
return ret_data
except UnavailableDataError as e:
if blocking:
- if db_name in inst.keyspace_notification_channels:
- result = inst._unavailable_data_handler(db_name, e.data)
+ if db_id in inst.keyspace_notification_channels:
+ result = inst._unavailable_data_handler(db_id, e.data)
if result:
continue # received updates, try to read data again
else:
- inst._unsubscribe_keyspace_notification(db_name)
+ inst._unsubscribe_keyspace_notification(db_id)
raise # No updates was received. Raise exception
else: # Subscribe to updates and try it again (avoiding race condition)
- inst._subscribe_keyspace_notification(db_name)
+ inst._subscribe_keyspace_notification(db_id)
else:
return None
except redis.exceptions.ResponseError:
@@ -57,12 +57,12 @@ def wrapped(inst, db_name, *args, **kwargs):
Retrying the request won't pass unless the schema itself changes. In this case, the error
should be attributed to the application itself. Re-raise the error.
"""
- logger.exception("Bad DB request [{}:{}]{{ {} }}".format(db_name, f.__name__, str(args)))
+ logger.exception("Bad DB request [{}:{}]{{ {} }}".format(db_id, f.__name__, str(args)))
raise
except (redis.exceptions.RedisError, OSError):
attempts += 1
- inst._connection_error_handler(db_name)
- msg = "DB access failure by [{}:{}]{{ {} }}".format(db_name, f.__name__, str(args))
+ inst._connection_error_handler(db_id)
+ msg = "DB access failure by [{}:{}]{{ {} }}".format(db_id, f.__name__, str(args))
if BLOCKING_ATTEMPT_ERROR_THRESHOLD < attempts < BLOCKING_ATTEMPT_SUPPRESSION:
# Repeated access failures implies the database itself is unhealthy.
logger.exception(msg=msg)
@@ -75,7 +75,7 @@ def wrapped(inst, db_name, *args, **kwargs):
class DBRegistry(dict):
def __getitem__(self, item):
if item not in self:
- raise MissingClientError("No client connected for db_name '{}'".format(item))
+ raise MissingClientError("No client connected for db_id '{}'".format(item))
return dict.__getitem__(self, item)
@@ -142,8 +142,6 @@ class DBInterface(object):
ACS Redis db mainly uses hash, therefore h is selected.
"""
- db_map = dict()
-
def __init__(self, **kwargs):
super(DBInterface, self).__init__()
@@ -161,147 +159,128 @@ def __init__(self, **kwargs):
# notifications for each client
self.keyspace_notification_channels = DBRegistry()
- for db_name in self.db_map:
- # set a database name as a constant value attribute.
- setattr(self, db_name, db_name)
-
- @property
- def db_list(self):
- return self.db_map.keys()
-
- @classmethod
- def get_dbid(cls, db_name):
- """
- :returns the database index by name. None if the name doesn't exist.
- """
- try:
- return cls.db_map[db_name]['db']
- except (KeyError, ValueError):
- return None
-
- def connect(self, db_name, retry_on=True):
+ def connect(self, db_id, retry_on=True):
"""
- :param db_name: named database to connect to
+ :param db_id: database id to connect to
:param retry_on: if ``True`` -- will attempt to connect continuously.
if ``False``, only one attempt will be made.
"""
if retry_on:
- self._persistent_connect(db_name)
+ self._persistent_connect(db_id)
else:
- self._onetime_connect(db_name)
+ self._onetime_connect(db_id)
- def _onetime_connect(self, db_name):
+ def _onetime_connect(self, db_id):
"""
- Connect to named database.
+ Connect to database id.
"""
- db_id = self.get_dbid(db_name)
if db_id is None:
- raise ValueError("No database ID configured for '{}'".format(db_name))
+ raise ValueError("No database ID configured for '{}'".format(db_id))
- client = redis.StrictRedis(db=self.db_map[db_name]['db'], **self.redis_kwargs)
+ client = redis.StrictRedis(db=db_id, **self.redis_kwargs)
# Enable the notification mechanism for keyspace events in Redis
client.config_set('notify-keyspace-events', self.KEYSPACE_EVENTS)
- self.redis_clients[db_name] = client
+ self.redis_clients[db_id] = client
- def _persistent_connect(self, db_name):
+ def _persistent_connect(self, db_id):
"""
- Keep reconnecting to Database 'db_name' until success
+ Keep reconnecting to Database 'db_id' until success
"""
while True:
try:
- self._onetime_connect(db_name)
+ self._onetime_connect(db_id)
return
except RedisError:
t_wait = self.CONNECT_RETRY_WAIT_TIME
- logger.warning("Connecting to DB '{}' failed, will retry in {}s".format(db_name, t_wait))
- self.close(db_name)
+ logger.warning("Connecting to DB '{}' failed, will retry in {}s".format(db_id, t_wait))
+ self.close(db_id)
time.sleep(t_wait)
- def close(self, db_name):
+ def close(self, db_id):
"""
Close all client(s) / keyspace channels.
- :param db_name: DB to disconnect from.
+ :param db_id: DB to disconnect from.
"""
- if db_name in self.redis_clients:
- self.redis_clients[db_name].connection_pool.disconnect()
- if db_name in self.keyspace_notification_channels:
- self.keyspace_notification_channels[db_name].close()
+ if db_id in self.redis_clients:
+ self.redis_clients[db_id].connection_pool.disconnect()
+ if db_id in self.keyspace_notification_channels:
+ self.keyspace_notification_channels[db_id].close()
- def _subscribe_keyspace_notification(self, db_name):
+ def _subscribe_keyspace_notification(self, db_id):
"""
Subscribe the chosent client to keyspace event notifications
"""
logger.debug("Subscribe to keyspace notification")
- client = self.redis_clients[db_name]
+ client = self.redis_clients[db_id]
pubsub = client.pubsub()
pubsub.psubscribe(self.KEYSPACE_PATTERN)
- self.keyspace_notification_channels[db_name] = pubsub
+ self.keyspace_notification_channels[db_id] = pubsub
- def _unsubscribe_keyspace_notification(self, db_name):
+ def _unsubscribe_keyspace_notification(self, db_id):
"""
Unsubscribe the chosent client from keyspace event notifications
"""
- if db_name in self.keyspace_notification_channels:
+ if db_id in self.keyspace_notification_channels:
logger.debug("Unsubscribe from keyspace notification")
- self.keyspace_notification_channels[db_name].close()
- del self.keyspace_notification_channels[db_name]
+ self.keyspace_notification_channels[db_id].close()
+ del self.keyspace_notification_channels[db_id]
- def get_redis_client(self, db_name):
+ def get_redis_client(self, db_id):
"""
- :param db_name: Name of the DB to query
+ :param db_id: Name of the DB to query
:return: The Redis client instance.
"""
- return self.redis_clients[db_name]
+ return self.redis_clients[db_id]
- def publish(self, db_name, channel, message):
+ def publish(self, db_id, channel, message):
"""
Publish message via the channel
"""
- client = self.redis_clients[db_name]
+ client = self.redis_clients[db_id]
return client.publish(channel, message)
- def expire(self, db_name, key, timeout_sec):
+ def expire(self, db_id, key, timeout_sec):
"""
Set a timeout on a key
"""
- client = self.redis_clients[db_name]
+ client = self.redis_clients[db_id]
return client.expire(key, timeout_sec)
- def exists(self, db_name, key):
+ def exists(self, db_id, key):
"""
Check if a key exist in the db
"""
- client = self.redis_clients[db_name]
+ client = self.redis_clients[db_id]
return client.exists(key)
@blockable
- def keys(self, db_name, pattern='*'):
+ def keys(self, db_id, pattern='*'):
"""
- Retrieve all the keys of DB %db_name
+ Retrieve all the keys of DB %db_id
"""
- client = self.redis_clients[db_name]
+ client = self.redis_clients[db_id]
keys = client.keys(pattern=pattern)
if not keys:
- message = "DB '{}' is empty!".format(db_name)
+ message = "DB '{}' is empty!".format(db_id)
logger.warning(message)
raise UnavailableDataError(message, b'hset')
else:
return keys
@blockable
- def get(self, db_name, _hash, key):
+ def get(self, db_id, _hash, key):
"""
Retrieve the value of Key %key from Hashtable %hash
- in Database %db_name
+ in Database %db_id
Parameter %blocking indicates whether to wait
when the query fails
"""
- client = self.redis_clients[db_name]
+ client = self.redis_clients[db_id]
val = client.hget(_hash, key)
if not val:
- message = "Key '{}' field '{}' unavailable in database '{}'".format(_hash, key, db_name)
+ message = "Key '{}' field '{}' unavailable in database '{}'".format(_hash, key, db_id)
logger.warning(message)
raise UnavailableDataError(message, _hash)
else:
@@ -309,17 +288,17 @@ def get(self, db_name, _hash, key):
return None if val == b'None' else val
@blockable
- def get_all(self, db_name, _hash):
+ def get_all(self, db_id, _hash):
"""
- Get Hashtable %hash from DB %db_name
+ Get Hashtable %hash from DB %db_id
Parameter %blocking indicates whether to wait
if the hashtable has not been created yet
"""
- client = self.redis_clients[db_name]
+ client = self.redis_clients[db_id]
table = client.hgetall(_hash)
if not table:
- message = "Key '{}' unavailable in database '{}'".format(_hash, db_name)
+ message = "Key '{}' unavailable in database '{}'".format(_hash, db_id)
logger.warning(message)
raise UnavailableDataError(message, _hash)
else:
@@ -327,35 +306,35 @@ def get_all(self, db_name, _hash):
return {k: None if v == b'None' else v for k, v in table.items()}
@blockable
- def set(self, db_name, _hash, key, val):
+ def set(self, db_id, _hash, key, val):
"""
- Add %(key, val) to Hashtable %hash in DB %db_name
+ Add %(key, val) to Hashtable %hash in DB %db_id
Parameter %blocking indicates whether to retry in case of failure
"""
- client = self.redis_clients[db_name]
+ client = self.redis_clients[db_id]
return client.hset(_hash, key, val)
@blockable
- def delete(self, db_name, key):
+ def delete(self, db_id, key):
"""
- Delete %key from DB %db_name
+ Delete %key from DB %db_id
Parameter %blocking indicates whether to retry in case of failure
"""
- client = self.redis_clients[db_name]
+ client = self.redis_clients[db_id]
return client.delete(key)
@blockable
- def delete_all_by_pattern(self, db_name, pattern):
+ def delete_all_by_pattern(self, db_id, pattern):
"""
- Delete all keys which match %pattern from DB %db_name
+ Delete all keys which match %pattern from DB %db_id
Parameter %blocking indicates whether to retry in case of failure
"""
- client = self.redis_clients[db_name]
+ client = self.redis_clients[db_id]
keys = client.keys(pattern)
for key in keys:
client.delete(key)
- def _unavailable_data_handler(self, db_name, data):
+ def _unavailable_data_handler(self, db_id, data):
"""
When the queried config is not available in Redis--wait until it is available.
Two timeouts are at work here:
@@ -363,23 +342,23 @@ def _unavailable_data_handler(self, db_name, data):
2. Max data wait - swsssdk-specific. how long to wait for the data to populate (in absolute time)
"""
start = time.time()
- logger.debug("Listening on pubsub channel '{}'".format(db_name))
+ logger.debug("Listening on pubsub channel '{}'".format(db_id))
while time.time() - start < self.PUB_SUB_MAXIMUM_DATA_WAIT:
- msg = self.keyspace_notification_channels[db_name].get_message(timeout=self.PUB_SUB_NOTIFICATION_TIMEOUT)
+ msg = self.keyspace_notification_channels[db_id].get_message(timeout=self.PUB_SUB_NOTIFICATION_TIMEOUT)
if msg is not None and msg.get('data') == data:
- logger.info("'{}' acquired via pub-sub. Unblocking...".format(data, db_name))
+ logger.info("'{}' acquired via pub-sub. Unblocking...".format(data, db_id))
# Wait for a "settling" period before releasing the wait.
time.sleep(self.DATA_RETRIEVAL_WAIT_TIME)
return True
- logger.warning("No notification for '{}' from '{}' received before timeout.".format(data, db_name))
+ logger.warning("No notification for '{}' from '{}' received before timeout.".format(data, db_id))
return False
- def _connection_error_handler(self, db_name):
+ def _connection_error_handler(self, db_id):
"""
In the event Redis is unavailable, close existing connections, and try again.
"""
logger.warning('Could not connect to Redis--waiting before trying again.')
- self.close(db_name)
+ self.close(db_id)
time.sleep(self.CONNECT_RETRY_WAIT_TIME)
- self.connect(db_name, True)
+ self.connect(db_id, True)
diff --git a/test/test_moduleLoad.py b/test/test_moduleLoad.py
index 870680f2e3da..0e81b15c0123 100644
--- a/test/test_moduleLoad.py
+++ b/test/test_moduleLoad.py
@@ -7,13 +7,9 @@
from unittest import TestCase
-class Test_load_connector_map(TestCase):
- def test__load_connector_map(self):
- # noinspection PyUnresolvedReferences
- import swsssdk
-
+class Test_load_sonic_db_config(TestCase):
def test__db_map_attributes(self):
import swsssdk
- db2 = swsssdk.SonicV2Connector()
- self.assertTrue(all(hasattr(db2, db_name) for db_name in db2.db_map))
+ db = swsssdk.SonicV2Connector()
+ self.assertTrue(all(hasattr(db, db_name) for db_name in db.get_db_list()))
pass