63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
from .database import *
|
|
|
|
|
|
def get_all_extension_data_keys(name: str) -> list[str]:
|
|
result: list[str] = []
|
|
for row in read_all(name, 'select key from extension_data'):
|
|
result.append(row['key'])
|
|
return result
|
|
|
|
|
|
def get_all_extension_data(name: str) -> dict[str, Any]:
|
|
result: dict[str, Any] = {}
|
|
for row in read_all(name, 'select key, value from extension_data'):
|
|
result[row['key']] = row['value']
|
|
return result
|
|
|
|
|
|
def get_extension_data(name: str, key: str) -> str | None:
|
|
if key == None or key == '':
|
|
return None
|
|
row = try_read(name, f"select value from extension_data where key = '{key}'")
|
|
if row == None:
|
|
return None
|
|
return row['value']
|
|
|
|
|
|
def _set_extension_data(name: str, cs: ChangeSet) -> DbChangeSet:
|
|
op = cs.operations[0]
|
|
key, new_val = op['key'], op['value']
|
|
|
|
f_new_val = f"'{new_val}'" if new_val != None else 'null'
|
|
|
|
old_val = get_extension_data(name, key)
|
|
f_old_val = f"'{old_val}'" if old_val != None else 'null'
|
|
|
|
redo_sql = f"delete from extension_data where key = '{key}';"
|
|
if new_val != None:
|
|
redo_sql += f"insert into extension_data (key, value) values ('{key}', {f_new_val});"
|
|
|
|
undo_sql = f"delete from extension_data where key = '{key}';"
|
|
if old_val != None:
|
|
undo_sql += f"insert into extension_data (key, value) values ('{key}', {f_old_val});"
|
|
|
|
redo_cs = g_update_prefix | { 'type': 'extension_data', 'key': key, 'value': new_val }
|
|
undo_cs = g_update_prefix | { 'type': 'extension_data', 'key': key, 'value': old_val }
|
|
|
|
return DbChangeSet(redo_sql, undo_sql, [redo_cs], [undo_cs])
|
|
|
|
|
|
def set_extension_data(name: str, cs: ChangeSet) -> ChangeSet:
|
|
if len(cs.operations) != 1:
|
|
return ChangeSet()
|
|
|
|
op = cs.operations[0]
|
|
if 'key' not in op or 'value' not in op:
|
|
return ChangeSet()
|
|
|
|
key = op['key']
|
|
if key == None or key == '':
|
|
return ChangeSet()
|
|
|
|
return execute_command(name, _set_extension_data(name, cs))
|