naeel
cbfba0bc54
По рекомендации Sonnet: - schema.py — все DDL (documents, contracts, supplements, spec_rows, spec_history) - db.py — только транспорт (connect/query), DDL убран - app.py — минимальный: ensure_schema() + register_blueprint() - test_routes.py — подробные комментарии, исправлен импорт psysql
91 lines
3.6 KiB
Python
91 lines
3.6 KiB
Python
"""
|
|
test_routes.py — Тестовый слой (/test Blueprint).
|
|
|
|
Мост к БД извне для отладки и диагностики.
|
|
Не зависит от бизнес-слоёв, только от db.py.
|
|
|
|
Эндпоинты:
|
|
GET /test — статус БД + список доступных действий
|
|
POST /test status — только статус БД
|
|
POST /test createdb — создать базу данных (если не существует)
|
|
POST /test tables — список таблиц в public схеме
|
|
POST /test sql {...} — выполнить произвольный SQL
|
|
"""
|
|
|
|
import os
|
|
from psycopg2 import sql as psysql
|
|
from flask import Blueprint, jsonify, request
|
|
import db
|
|
|
|
test_bp = Blueprint("test", __name__)
|
|
|
|
|
|
@test_bp.route("/test", methods=["GET", "POST"])
|
|
def test():
|
|
"""
|
|
Тестовый слой: проверка БД и выполнение запросов.
|
|
GET — статус подключения
|
|
POST — выполнить действие (status/tables/sql/createdb)
|
|
"""
|
|
|
|
# ── GET — показать статус и список команд ────────────────────
|
|
if request.method == "GET":
|
|
conn, err = db.connect()
|
|
return jsonify({
|
|
"db": "ok" if conn else f"fail: {err}",
|
|
"actions": [
|
|
"GET /test — статус БД",
|
|
"POST /test status — статус БД (JSON)",
|
|
"POST /test createdb — создать БД (если нет)",
|
|
"POST /test tables — список таблиц",
|
|
"POST /test sql {...} — выполнить SQL",
|
|
],
|
|
})
|
|
|
|
# ── POST — выполнить действие ────────────────────────────────
|
|
data = request.get_json(silent=True) or {}
|
|
action = data.get("action", "status")
|
|
|
|
if action == "status":
|
|
conn, err = db.connect()
|
|
return jsonify({"db": "ok" if conn else f"fail: {err}"})
|
|
|
|
if action == "tables":
|
|
result, err = db.query(
|
|
"SELECT table_name FROM information_schema.tables "
|
|
"WHERE table_schema = 'public' ORDER BY table_name"
|
|
)
|
|
if err:
|
|
return jsonify({"error": err}), 500
|
|
return jsonify({"tables": [r[0] for r in result["rows"]]})
|
|
|
|
if action == "createdb":
|
|
# Подключаемся к дефолтной БД 'postgres' и создаём целевую
|
|
try:
|
|
conn = db._pg_connect("postgres")
|
|
conn.autocommit = True
|
|
cur = conn.cursor()
|
|
db_name = data.get("name") or os.environ.get("DB_NAME", "contracts")
|
|
cur.execute("SELECT 1 FROM pg_database WHERE datname = %s", (db_name,))
|
|
if cur.fetchone():
|
|
msg = f"DB '{db_name}' already exists"
|
|
else:
|
|
cur.execute(psysql.SQL("CREATE DATABASE {}").format(psysql.Identifier(db_name)))
|
|
msg = f"DB '{db_name}' created"
|
|
cur.close()
|
|
conn.close()
|
|
return jsonify({"result": msg})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
if action == "sql":
|
|
sql_text = data.get("sql", "")
|
|
if not sql_text:
|
|
return jsonify({"error": "no sql"}), 400
|
|
result, err = db.query(sql_text)
|
|
if err:
|
|
return jsonify({"error": err}), 500
|
|
return jsonify(result)
|
|
|
|
return jsonify({"error": f"unknown action: {action}"}), 400
|