Add status api and test
This commit is contained in:
@@ -57,6 +57,12 @@ def read_all(name: str, sql: str) -> list[Row]:
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
def try_read(name: str, sql: str) -> Row | None:
|
||||
with conn[name].cursor(row_factory=dict_row) as cur:
|
||||
cur.execute(sql)
|
||||
return cur.fetchone()
|
||||
|
||||
|
||||
def write(name: str, sql: str) -> None:
|
||||
with conn[name].cursor() as cur:
|
||||
cur.execute(sql)
|
||||
|
||||
Reference in New Issue
Block a user