Base is readonly

api/s0_base.py

@@ -1,6 +1,5 @@
-from psycopg.rows import dict_row
+from psycopg.rows import dict_row, Row
 from connection import g_conn_dict as conn
-from operation import *
 
 _NODE = "_node"
 _LINK = "_link"
@@ -14,151 +13,43 @@ PIPE = "PIPE"
 PUMP = "PUMP"
 VALVE = "VALVE"
 
-nodeTables = { JUNCTION: "junctions", RESERVOIR: "reservoirs", TANK: "tanks" }
+def _get_from(name: str, id: str, base_type: str) -> Row | None:
-linkTables = { PIPE: "pipes", PUMP: "pumps", VALVE: "valves" }
-typeTales = { _NODE: nodeTables, _LINK: linkTables }
-
-# add
-
-def _add_id_type(name: str, id: str, type: str, base_type: str) -> None:
-    with conn[name].cursor() as cur:
-        sql = f"insert into {base_type} (id, type) values ('{id}', '{type}'); insert into {typeTales[base_type][type]} (id) values ('{id}');"
-        if base_type == _NODE:
-            sql += f" insert into coordinates (node) values ('{id}');"
-        cur.execute(sql)
-
-        redo = sql.replace("'", '"')
-        undo = ""
-        if base_type == _NODE:
-            undo = f'delete from coordinates where node = "{id}"; '
-        undo += f'delete from {typeTales[base_type][type]} where id = "{id}"; delete from {base_type} where id = "{id}";'
-        add_operation(name, redo, undo)
-
-def _add_id(name: str, id: str, base_type: str) -> None:
-    with conn[name].cursor() as cur:
-        sql = f"insert into {base_type} (id) values ('{id}')"
-        cur.execute(sql)
-        redo = sql.replace("'", '"')
-        undo = f'delete from {base_type} where id = "{id}"'
-        add_operation(name, redo, undo)
-
-def add_node(name: str, id: str, type: str) -> None:
-    return _add_id_type(name, id, type, _NODE)
-
-def add_link(name: str, id: str, type: str) -> None:
-    return _add_id_type(name, id, type, _LINK)
-
-def add_curve(name: str, id: str) -> None:
-    return _add_id(name, id, _CURVE)
-
-def add_pattern(name: str, id: str) -> None:
-    return _add_id(name, id, _PATTERN)
-
-
-# have
-
-def _have_impl(name: str, id: str, base_type: str) -> bool:
-    with conn[name].cursor() as cur:
-        cur.execute(f"select * from {base_type} where id = '{id}'")
-        return cur.rowcount > 0
-
-def have_node(name: str, id: str) -> bool:
-    return _have_impl(name, id, _NODE)
-
-def have_link(name: str, id: str) -> bool:
-    return _have_impl(name, id, _LINK)
-
-def have_curve(name: str, id: str) -> bool:
-    return _have_impl(name, id, _CURVE)
-
-def have_pattern(name: str, id: str) -> bool:
-    return _have_impl(name, id, _PATTERN)
-
-
-# get
-def _get_impl(name: str, id: str, base_type: str) -> dict[str, str]:
     with conn[name].cursor(row_factory=dict_row) as cur:
         cur.execute(f"select * from {base_type} where id = '{id}'")
-        if cur.rowcount > 0:
+        return cur.fetchone()
-            return cur.fetchone()
-        else:
-            return {}
 
-def get_node(name: str, id: str) -> dict[str, str]:
+def is_node(name: str, id: str) -> bool:
-    return _get_impl(name, id, _NODE)
+    return _get_from(name, id, _NODE) != None
 
 def is_junction(name: str, id: str) -> bool:
-    return get_node(name, id)['type'] == JUNCTION
+    row = _get_from(name, id, _NODE)
+    return row != None and row['type'] == JUNCTION
 
 def is_reservoir(name: str, id: str) -> bool:
-    return get_node(name, id)['type'] == RESERVOIR
+    row = _get_from(name, id, _NODE)
+    return row != None and row['type'] == RESERVOIR
 
 def is_tank(name: str, id: str) -> bool:
-    return get_node(name, id)['type'] == TANK
+    row = _get_from(name, id, _NODE)
+    return row != None and row['type'] == TANK
 
-def get_link(name: str, id: str) -> dict[str, str]:
+def is_link(name: str, id: str) -> bool:
-    return _get_impl(name, id, _LINK)
+    return _get_from(name, id, _LINK) != {}
 
 def is_pipe(name: str, id: str) -> bool:
-    return get_link(name, id)['type'] == PIPE
+    row = _get_from(name, id, _LINK)
+    return row != None and row['type'] == PIPE
 
 def is_pump(name: str, id: str) -> bool:
-    return get_link(name, id)['type'] == PUMP
+    row = _get_from(name, id, _LINK)
+    return row != None and row['type'] == PUMP
 
 def is_valve(name: str, id: str) -> bool:
-    return get_link(name, id)['type'] == VALVE
+    row = _get_from(name, id, _LINK)
+    return row != None and row['type'] == VALVE
 
-def get_curve(name: str, id: str) -> dict[str, str]:
+def is_curve(name: str, id: str) -> bool:
-    return _get_impl(name, id, _CURVE)
+    return _get_from(name, id, _CURVE) != None
 
-def get_pattern(name: str, id: str) -> dict[str, str]:
+def is_pattern(name: str, id: str) -> bool:
-    return _get_impl(name, id, _PATTERN)
+    return _get_from(name, id, _PATTERN) != None
-
-
-# delete
-
-def _delete_id_type(name: str, id: str, type: str, base_type: str) -> None:
-    with conn[name].cursor() as cur:
-        sql = ""
-        if base_type == _NODE:
-            sql = f"delete from coordinates where node = '{id}'; "
-        sql += f"delete from {typeTales[base_type][type]} where id = '{id}'; delete from {base_type} where id = '{id}';"
-        cur.execute(sql)
-
-        redo = sql.replace("'", '"')
-        undo = f'insert into {base_type} (id, type) values ("{id}", "{type}"); insert into {typeTales[base_type][type]} (id) values ("{id}");'
-        if base_type == _NODE:
-            undo += f' insert into coordinates (node) values ("{id}");'
-        add_operation(name, redo, undo)
-
-def _delete_id(name:str, id: str, base_type: str) -> None:
-    with conn[name].cursor() as cur:
-        sql = f"delete from {base_type} where id = '{id}'"
-        cur.execute(sql)
-        redo = sql.replace("'", '"')
-        undo = f'insert into {base_type} (id) values ("{id}")'
-        add_operation(name, redo, undo)
-
-def delete_node(name: str, id: str) -> None:
-    row = get_node(name, id)
-    if row == {}:
-        return
-    return _delete_id_type(name, id, row['type'], _NODE)
-
-def delete_link(name: str, id: str) -> None:
-    row = get_link(name, id)
-    if row == {}:
-        return
-    return _delete_id_type(name, id, row['type'], _LINK)
-
-def delete_curve(name: str, id: str) -> None:
-    row = get_curve(name, id)
-    if row == {}:
-        return
-    return _delete_id(name, id, _CURVE)
-
-def delete_pattern(name: str, id: str) -> None:
-    row = get_pattern(name, id)
-    if row == {}:
-        return
-    return _delete_id(name, id, _PATTERN)