Read Data from database table and access it as Tuple
Call the method using
self.data_movement_config, self._data_movement_name = \
self._read_config()
it will call the read_config method and the returned tuple will be saved to two variables , data_movement_config, data_movement_name
def _read_config(self) -> Tuple[Dict, str]:
"""Read configuration from SupportDB .
Returns:
A tuple with the dictionary of the JSON configuration for a given
and the data movement name.
sql_query = f"""
SELECT
ConfigurationJSON,
DataMovementName
FROM
ctrl.DataMovement
WHERE DataMovementID = '{self._data_movement_id}'
"""
try:
cursor = self._support_db_conn.execute_with_retry(sql_query)
except: # noqa E722
raise
db_row = cursor.fetchone()
if db_row is None:
raise ValueError("The DataMovementID did not return any rows.")
try:
return json.loads(db_row[0]), db_row[1] # return two fields , as db_row[0]) as json , json_loads
except (JSONDecodeError, json.JSONDecodeError, TypeError):
raise ValueError(
"The DataMovementID did not return a valid JSON configuration.")
Comments
Post a Comment