feat: слои db.py + test_routes (Blueprint /test)

This commit is contained in:
2026-06-13 12:02:46 +04:00
parent 684b4cc9c0
commit dbbc8ea5fe
4 changed files with 89 additions and 18 deletions
+36
View File
@@ -0,0 +1,36 @@
import os
import psycopg2
from psycopg2 import sql
def connect():
"""Подключение к БД. Возвращает connection или None."""
try:
return psycopg2.connect(
host=os.getenv("DB_HOST"),
port=os.getenv("DB_PORT", 5432),
dbname=os.getenv("DB_NAME"),
user=os.getenv("DB_USER"),
password=os.getenv("DB_PASS"),
sslmode=os.getenv("DB_SSLMODE", "disable"),
)
except Exception:
return None
def query(sql_text, params=None):
"""Выполнить запрос, вернуть (rows, columns) или (None, error)."""
conn = connect()
if not conn:
return None, "no connection"
try:
cur = conn.cursor()
cur.execute(sql_text, params)
rows = cur.fetchall()
cols = [desc[0] for desc in cur.description] if cur.description else []
cur.close()
conn.close()
return {"columns": cols, "rows": [list(r) for r in rows]}, None
except Exception as e:
conn.close()
return None, str(e)