34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
from psycopg.rows import dict_row, Row
|
|
from .connection import g_conn_dict as conn
|
|
from .operation import *
|
|
from .change_set import ChangeSet
|
|
|
|
|
|
def query(name: str, sql: str) -> Row | None:
|
|
with conn[name].cursor(row_factory=dict_row) as cur:
|
|
cur.execute(sql)
|
|
return cur.fetchone()
|
|
|
|
|
|
def decorate(value: str | None, type: str, optional: bool) -> str:
|
|
if optional:
|
|
value = 'NULL' if value == None else value
|
|
if type == 'str':
|
|
value = f'"{value}"' if value != 'NULL' else value
|
|
return value
|
|
|
|
|
|
def update(name: str, type: str, table: str, id_key: str, id_value: str, key: str, key_type: str, raw_old_value: str, value: str, optional: bool = False) -> ChangeSet:
|
|
old = decorate(raw_old_value, key_type, optional)
|
|
|
|
with conn[name].cursor() as cur:
|
|
sql = f"update {table} set {key} = {value} where {id_key} = '{id_value}'"
|
|
cur.execute(sql)
|
|
redo = sql.replace("'", '"')
|
|
undo = f'update {table} set {key} = {old} where {id_key} = "{id_value}"'
|
|
add_operation(name, redo, undo)
|
|
|
|
change = ChangeSet()
|
|
change.update(type, id, key, key_type, value)
|
|
return change
|