From 9b2b54e9abd9d195f91ad46680f061ce2a24d557 Mon Sep 17 00:00:00 2001 From: wqy Date: Sat, 3 Sep 2022 00:33:58 +0800 Subject: [PATCH] Base is readonly --- api/s0_base.py | 155 ++++++++----------------------------------------- 1 file changed, 23 insertions(+), 132 deletions(-) diff --git a/api/s0_base.py b/api/s0_base.py index 83be9fd..bbefa7b 100644 --- a/api/s0_base.py +++ b/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" } -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]: +def _get_from(name: str, id: str, base_type: str) -> Row | None: 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() - else: - return {} + return cur.fetchone() -def get_node(name: str, id: str) -> dict[str, str]: - return _get_impl(name, id, _NODE) +def is_node(name: str, id: str) -> bool: + 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]: - return _get_impl(name, id, _LINK) +def is_link(name: str, id: str) -> bool: + 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]: - return _get_impl(name, id, _CURVE) +def is_curve(name: str, id: str) -> bool: + return _get_from(name, id, _CURVE) != None -def get_pattern(name: str, id: str) -> dict[str, str]: - return _get_impl(name, id, _PATTERN) - - -# 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) +def is_pattern(name: str, id: str) -> bool: + return _get_from(name, id, _PATTERN) != None