feat: schema.py (DDL 5 таблиц), db.py без DDL, app.py минимальный, test_routes с комментариями

По рекомендации 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
This commit is contained in:
2026-06-13 17:14:48 +04:00
parent 308be45ce8
commit cbfba0bc54
5 changed files with 260 additions and 38 deletions
+29 -7
View File
@@ -1,4 +1,19 @@
"""
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
@@ -7,22 +22,27 @@ test_bp = Blueprint("test", __name__)
@test_bp.route("/test", methods=["GET", "POST"])
def test():
"""Тестовый слой: проверка БД и выполнение запросов."""
"""
Тестовый слой: проверка БД и выполнение запросов.
GET — статус подключения
POST — выполнить действие (status/tables/sql/createdb)
"""
# GET — статус
# ── GET — показать статус и список команд ────────────────────
if request.method == "GET":
conn, err = db.connect()
return jsonify({
"db": "ok" if conn else f"fail: {err}",
"actions": [
"GET /test — статус БД",
"POST /test createdb — создать БД",
"POST /test status — статус БД (JSON)",
"POST /test createdb — создать БД (если нет)",
"POST /test tables — список таблиц",
"POST /test SQL:... — выполнить запрос",
"POST /test sql {...} — выполнить SQL",
],
})
# POST — выполнить действие
# ── POST — выполнить действие ────────────────────────────────
data = request.get_json(silent=True) or {}
action = data.get("action", "status")
@@ -32,13 +52,15 @@ def test():
if action == "tables":
result, err = db.query(
"SELECT table_name FROM information_schema.tables WHERE table_schema='public' ORDER BY table_name"
"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
@@ -48,7 +70,7 @@ def test():
if cur.fetchone():
msg = f"DB '{db_name}' already exists"
else:
cur.execute(db.sql.SQL("CREATE DATABASE {}").format(db.sql.Identifier(db_name)))
cur.execute(psysql.SQL("CREATE DATABASE {}").format(psysql.Identifier(db_name)))
msg = f"DB '{db_name}' created"
cur.close()
conn.close()