79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""db_bridge_vm.py – VM strana souborového mostu k Medicusu.
|
||
|
||
Použití z Linuxu:
|
||
from db_bridge_vm import query
|
||
rows, columns = query("SELECT COUNT(*) FROM KAR")
|
||
print(columns, rows)
|
||
"""
|
||
import json, time, os, uuid
|
||
|
||
BRIDGE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
REQUEST = os.path.join(BRIDGE_DIR, 'query_request.json')
|
||
RESPONSE = os.path.join(BRIDGE_DIR, 'query_response.json')
|
||
|
||
TIMEOUT_SEC = 30
|
||
POLL_SEC = 0.3
|
||
|
||
|
||
def query(sql, params=None, timeout=TIMEOUT_SEC):
|
||
"""Pošle SQL dotaz přes souborový most a vrátí (rows, columns).
|
||
|
||
Raises:
|
||
TimeoutError – watchdog neodpověděl do timeout sekund
|
||
RuntimeError – Firebird vrátil chybu
|
||
"""
|
||
# Smaž případnou starou response
|
||
if os.path.exists(RESPONSE):
|
||
os.remove(RESPONSE)
|
||
|
||
req_id = uuid.uuid4().hex
|
||
req = {'id': req_id, 'sql': sql, 'params': params or []}
|
||
|
||
with open(REQUEST, 'w', encoding='utf-8') as f:
|
||
json.dump(req, f, ensure_ascii=False)
|
||
|
||
# Čekej na odpověď
|
||
waited = 0.0
|
||
while waited < timeout:
|
||
time.sleep(POLL_SEC)
|
||
waited += POLL_SEC
|
||
if os.path.exists(RESPONSE):
|
||
with open(RESPONSE, 'r', encoding='utf-8') as f:
|
||
resp = json.load(f)
|
||
os.remove(RESPONSE)
|
||
if resp.get('status') == 'error':
|
||
raise RuntimeError(f"DB chyba: {resp.get('error')}")
|
||
return resp.get('rows', []), resp.get('columns', [])
|
||
|
||
# Timeout – smaž request aby watchdog nezpracoval zastaralý dotaz
|
||
if os.path.exists(REQUEST):
|
||
os.remove(REQUEST)
|
||
raise TimeoutError(f'Watchdog neodpověděl do {timeout}s – běží db_bridge_windows.py?')
|
||
|
||
|
||
def query_print(sql, params=None):
|
||
"""Spustí dotaz a vypíše výsledek přehledně."""
|
||
rows, cols = query(sql, params)
|
||
if not cols:
|
||
print('(žádné sloupce)')
|
||
return rows, cols
|
||
col_w = [max(len(str(c)), max((len(str(r[i])) for r in rows), default=0))
|
||
for i, c in enumerate(cols)]
|
||
sep = '+' + '+'.join('-' * (w + 2) for w in col_w) + '+'
|
||
fmt = '|' + '|'.join(f' {{:<{w}}} ' for w in col_w) + '|'
|
||
print(sep)
|
||
print(fmt.format(*cols))
|
||
print(sep)
|
||
for row in rows:
|
||
print(fmt.format(*[str(v) if v is not None else 'NULL' for v in row]))
|
||
print(sep)
|
||
print(f'{len(rows)} řádků')
|
||
return rows, cols
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# Rychlý test
|
||
print('Testuji spojení...')
|
||
rows, cols = query('SELECT COUNT(*) AS POCET FROM KAR')
|
||
print(f'OK – pacientů v KAR: {rows[0][0]}')
|