I'm doing something similar to that for storing some trend related properties per user. You're welcome to try and decipher and see if this helps you. This packages the config into a message handler, sends it to a remote gateway, and executes a namedquery to save/load it on the other gateway's database.
Fair warning - AI supported coding, so may not be best practices but it currently works well for me.
def savequickpoints(UserName, Data):
"""
Saves the quickpoints dataset to the database as a JSON string.
"""
# 1. Build the list from the dataset
points_list = []
if Data is not None:
py_data = system.dataset.toPyDataSet(Data)
for row in py_data:
points_list.append({
"Pad": row["Pad"],
"Pen Name": row["Pen Name"],
"Description": row["Description"],
"Path": row["Path"]
})
# 2. Encode to JSON
# THIS IS THE CRITICAL STEP.
# We must convert the list of objects into a text string.
json_string = system.util.jsonEncode(points_list)
# DEBUG: Print this to the console to confirm it looks like text, not "Dataset [...]"
print "Saving the following string to SQL:", json_string
# 3. Call the remote data handler
return get_remote_data(
query_path="custom_trend/savequickpoints",
database="-redacted-",
UserName = UserName,
Data = json_string
)
def loadquickpoints(UserName):
"""
Fetches the JSON string from SQL and converts it to a Dataset.
"""
# 1. Fetch the data
result = get_remote_data(
query_path="custom_trend/loadquickpoints",
database="-redacted-",
UserName = UserName
)
# 2. Sanitize the Result (Handle Dataset vs String)
json_string = ""
if result is None:
return None
# Check if result is an Ignition Dataset (common if Query Type is 'Select')
if hasattr(result, 'getValueAt'):
if result.getRowCount() > 0:
json_string = result.getValueAt(0, 0)
else:
return None
else:
# Assume it is already the scalar string
json_string = result
# 3. Decode JSON
if not json_string:
return None
try:
data_structure = system.util.jsonDecode(json_string)
except:
# Log the error if needed, or just return None
return None
# 4. Normalize to List
# This prevents the 'unicode' error by ensuring we are looping over a list of objects,
# not iterating over the keys of a single dictionary.
if isinstance(data_structure, dict):
data_list = [data_structure]
elif isinstance(data_structure, list):
data_list = data_structure
else:
return None # Unknown format
# 5. Convert to Dataset
headers = ["Pad", "Pen Name", "Description", "Path"]
rows = []
for item in data_list:
# Critical Check: ensure the item is a dictionary before calling .get()
if isinstance(item, dict):
rows.append([
item.get("Pad"),
item.get("Pen Name"),
item.get("Description"),
item.get("Path")
])
return system.dataset.toDataSet(headers, rows)
The named queries are just:
loadquickpoints:
SELECT Data
FROM custom_quickpoints
WHERE UserName=:UserName
savequickpoints:
IF EXISTS (SELECT 1 FROM custom_quickpoints WHERE UserName = :UserName)
BEGIN
-- User exists: Update their existing record
UPDATE custom_quickpoints
SET Data = :Data
WHERE UserName = :UserName
END
ELSE
BEGIN
-- New User: Insert a new record
INSERT INTO custom_quickpoints (UserName, Data)
VALUES (:UserName, :Data)
END
Edit: The on change script that is attached to session.custom.quickpoints:
def valueChanged(self, previousValue, currentValue, origin, missedEvents):
UserName = self.props.auth.user.userName
Data = self.custom.quickpoints
if UserName:
remote_utils.savequickpoints(UserName, Data)