OrgTJWater/TJWaterServer
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix module in terms of python style

Browse Source
This commit is contained in:
wqy
2022-09-02 18:33:33 +08:00
parent 4474f12d2c
commit bed5741558
11 changed files with 214 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
api/_0_base.py
View File

@@ -1,94 +0,0 @@
from _connection import _conn_dict as conn
from psycopg.rows import dict_row
_NODE = "_NODE"
_LINK = "_LINK"
_CURVE = "_CURVE"
_PATTERN = "_PATTERN"
JUNCTION = "JUNCTION"
RESERVOIR = "RESERVOIR"
TANK = "TANK"
PIPE = "PIPE"
PUMP = "PUMP"
VALVE = "VALVE"
# add
def add_node(name: str, id: str, type: str) -> None:
with conn[name].cursor() as cur:
cur.execute(f"INSERT INTO _NODE (ID, Type) VALUES ('{id}', '{type}')")
def add_link(name: str, id: str, type: str) -> None:
with conn[name].cursor() as cur:
cur.execute(f"INSERT INTO _LINK (ID, Type) VALUES ('{id}', '{type}')")
def add_curve(name: str, id: str) -> None:
with conn[name].cursor() as cur:
cur.execute(f"INSERT INTO _CURVE (ID) VALUES ('{id}')")
def add_pattern(name: str, id: str) -> None:
with conn[name].cursor() as cur:
cur.execute(f"INSERT INTO _PATTERN (ID) VALUES ('{id}')")
# have
def _have_impl(name: str, id: str, table: str) -> bool:
with conn[name].cursor() as cur:
cur.execute(f"SELECT * FROM {table} 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, table: str) -> dict[str, str]:
with conn[name].cursor(row_factory=dict_row) as cur:
cur.execute(f"SELECT * FROM {table} WHERE ID = '{id}'")
if cur.rowcount > 0:
return cur.fetchone()
else:
return {}
def get_node(name: str, id: str) -> dict[str, str]:
return _get_impl(name, id, _NODE)
def get_link(name: str, id: str) -> dict[str, str]:
return _get_impl(name, id, _LINK)
def get_curve(name: str, id: str) -> dict[str, str]:
return _get_impl(name, id, _CURVE)
def get_pattern(name: str, id: str) -> dict[str, str]:
return _get_impl(name, id, _PATTERN)
# delete
def _delete_impl(name: str, id: str, table: str) -> None:
with conn[name].cursor() as cur:
cur.execute(f"DELETE FROM {table} WHERE ID = '{id}'")
def delete_node(name: str, id: str) -> None:
return _delete_impl(name, id, _NODE)
def delete_link(name: str, id: str) -> None:
return _delete_impl(name, id, _LINK)
def delete_curve(name: str, id: str) -> None:
return _delete_impl(name, id, _CURVE)
def delete_pattern(name: str, id: str) -> None:
return _delete_impl(name, id, _PATTERN)

30
api/_0_base_test.py
View File

@@ -1,30 +0,0 @@
from _project import *
from _0_base import *
open_project("net")
add_node("net", "n-1", JUNCTION)
print(get_node("net", "n-1"))
if have_node("net", "n-1"):
delete_node("net", "n-1")
print(have_node("net", "n-1"))
add_link("net", "l-1", PIPE)
print(get_link("net", "l-1"))
if have_link("net", "l-1"):
delete_link("net", "l-1")
print(have_link("net", "l-1"))
add_curve("net", "c-1")
print(get_curve("net", "c-1"))
if have_curve("net", "c-1"):
delete_curve("net", "c-1")
print(have_curve("net", "c-1"))
add_pattern("net", "p-1")
print(get_pattern("net", "p-1"))
if have_pattern("net", "p-1"):
delete_pattern("net", "p-1")
print(have_pattern("net", "p-1"))
close_project("net")

28
api/_1_title.py
View File

@@ -1,28 +0,0 @@
from _connection import _conn_dict as conn
from psycopg.rows import dict_row
def have_title(name: str) -> bool:
with conn[name].cursor() as cur:
cur.execute(f"SELECT * FROM TITLE")
return cur.rowcount > 0
def set_title(name: str, value: str) -> None:
if have_title(name):
with conn[name].cursor() as cur:
cur.execute(f"UPDATE TITLE SET Value = '{value}'")
else:
with conn[name].cursor() as cur:
cur.execute(f"INSERT INTO TITLE (Value) VALUES ('{value}')")
def get_title(name: str) -> str:
with conn[name].cursor(row_factory=dict_row) as cur:
cur.execute(f"SELECT * FROM TITLE")
if cur.rowcount > 0:
return cur.fetchone()['value']
else:
return ""
def unset_title(name: str) -> None:
with conn[name].cursor() as cur:
cur.execute(f"TRUNCATE TITLE")
return cur.rowcount > 0

29
api/_1_title_test.py
View File

@@ -1,29 +0,0 @@
from _project import *
from _1_title import *
open_project("net")
print(have_title("net"))
print(get_title("net"))
set_title("net", "xxx")
print(have_title("net"))
print(get_title("net"))
set_title("net", "xxxx")
print(have_title("net"))
print(get_title("net"))
unset_title("net")
print(have_title("net"))
print(get_title("net"))
set_title("net", "xxxx")
print(have_title("net"))
print(get_title("net"))
unset_title("net")
print(have_title("net"))
print(get_title("net"))
close_project("net")

3
api/_connection.py
View File

@@ -1,3 +0,0 @@
import psycopg as pg
_conn_dict : dict[str, pg.Connection] = {}

3
api/connection.py Normal file
View File

@@ -0,0 +1,3 @@
import psycopg as pg
g_conn_dict : dict[str, pg.Connection] = {}

2
api/_project.py → api/project.py
View File

@@ -1,5 +1,5 @@
import psycopg as pg
from _connection import _conn_dict as conn
from api.connection import g_conn_dict as conn
def have_project(name: str) -> bool:
with pg.connect(conninfo="dbname=postgres", autocommit=True) as conn:

131
api/s0_base.py Normal file
View File

@@ -0,0 +1,131 @@
from psycopg.rows import dict_row
from api.connection import g_conn_dict as conn
from api.operation import *
_NODE = "_node"
_LINK = "_link"
_CURVE = "_curve"
_PATTERN = "_pattern"
JUNCTION = "JUNCTION"
RESERVOIR = "RESERVOIR"
TANK = "TANK"
PIPE = "PIPE"
PUMP = "PUMP"
VALVE = "VALVE"
# add
def _add_id_type(name: str, id: str, type: str, table: str) -> None:
with conn[name].cursor() as cur:
sql = f"insert into {table} (id, type) values ('{id}', '{type}')"
cur.execute(sql)
redo = sql.replace("'", '"')
undo = f'delete from {table} where id = "{id}"'
add_operation(name, redo, undo)
def _add_id(name: str, id: str, table: str) -> None:
with conn[name].cursor() as cur:
sql = f"insert into {table} (id) values ('{id}')"
cur.execute(sql)
redo = sql.replace("'", '"')
undo = f'delete from {table} 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, table: str) -> bool:
with conn[name].cursor() as cur:
cur.execute(f"select * from {table} 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, table: str) -> dict[str, str]:
with conn[name].cursor(row_factory=dict_row) as cur:
cur.execute(f"select * from {table} where id = '{id}'")
if cur.rowcount > 0:
return cur.fetchone()
else:
return {}
def get_node(name: str, id: str) -> dict[str, str]:
return _get_impl(name, id, _NODE)
def get_link(name: str, id: str) -> dict[str, str]:
return _get_impl(name, id, _LINK)
def get_curve(name: str, id: str) -> dict[str, str]:
return _get_impl(name, id, _CURVE)
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, table: str) -> None:
with conn[name].cursor() as cur:
sql = f"delete from {table} where id = '{id}'"
cur.execute(sql)
redo = sql.replace("'", '"')
undo = f'insert into {table} (id, type) values ("{id}", "{type}")'
add_operation(name, redo, undo)
def _delete_id(name:str, id: str, table: str) -> None:
with conn[name].cursor() as cur:
sql = f"delete from {table} where id = '{id}'"
cur.execute(sql)
redo = sql.replace("'", '"')
undo = f'insert into {table} (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)

11
api/s1_title.py Normal file
View File

@@ -0,0 +1,11 @@
from psycopg.rows import dict_row
from api.connection import g_conn_dict as conn
def set_title(name: str, value: str) -> None:
with conn[name].cursor() as cur:
cur.execute(f"update title set value = '{value}'")
def get_title(name: str) -> str:
with conn[name].cursor(row_factory=dict_row) as cur:
cur.execute(f"select * from title")
return cur.fetchone()['value']

44
api/test/s0_base.py Normal file
View File

@@ -0,0 +1,44 @@
from api.project import *
from api.s0_base import *
p = "test_s0_base"
n_1 = "n_1"
l_1 = "l_1"
c_1 = "c_1"
p_1 = "p_1"
if is_project_open(p):
close_project(p)
if have_project(p):
delete_project(p)
create_project(p)
open_project(p)
add_node(p, n_1, JUNCTION)
print(get_node(p, n_1))
if have_node(p, n_1):
delete_node(p, n_1)
print(have_node(p, n_1))
add_link(p, l_1, PIPE)
print(get_link(p, l_1))
if have_link(p, l_1):
delete_link(p, l_1)
print(have_link(p, l_1))
add_curve(p, c_1)
print(get_curve(p, c_1))
if have_curve(p, c_1):
delete_curve(p, c_1)
print(have_curve(p, c_1))
add_pattern(p, p_1)
print(get_pattern(p, p_1))
if have_pattern(p, p_1):
delete_pattern(p, p_1)
print(have_pattern(p, p_1))
close_project(p)
delete_project(p)

24
api/test/s1_title.py Normal file
View File

@@ -0,0 +1,24 @@
from api.project import *
from api.s1_title import *
p = "test_s1_title"
if is_project_open(p):
close_project(p)
if have_project(p):
delete_project(p)
create_project(p)
open_project(p)
print(get_title(p))
set_title(p, "title")
print(get_title(p))
set_title(p, "test")
print(get_title(p))
close_project(p)
delete_project(p)