125 lines
4.6 KiB
Python
125 lines
4.6 KiB
Python
"""把插槽视图 DDL 应用到基座 PG —— 用桥自己的 psycopg 连接，不依赖 psql 客户端。
|
||
|
||
**为什么需要它**：桥是跨机部署的独立单元（设计 §4），部署它的服务器上通常
|
||
**没有** astock-kg 的 `akg-postgres` 容器，README 里那条
|
||
`docker exec -i akg-postgres psql ...` 只在基座那台机器上成立。
|
||
而桥容器是 python:3.11-slim，也没有 psql 客户端。
|
||
本模块用桥已有的 `AKG_PG_*` 连接执行 DDL，零新增凭据、全程 Docker。
|
||
|
||
**边界说明**：视图定义文件本来就随桥仓库版本控制（sql/astock_kg_slot_views.sql），
|
||
所以由桥来应用它并不越界——桥定义插槽接口，基座只是接口的宿主。
|
||
但注意**权限**：日常运行只需只读账号，应用 DDL 需要能 CREATE VIEW / ALTER TABLE
|
||
的账号（通常是基座 owner）。若报 permission denied，用 owner 账号跑一次即可：
|
||
|
||
docker compose exec -T \
|
||
-e AKG_PG_USER=<owner> -e AKG_PG_PASSWORD=<pw> \
|
||
akg-factor-bridge python run.py apply-views
|
||
|
||
（注意必须用 `docker compose exec -e`：在命令前面写 `AKG_PG_USER=... docker compose exec`
|
||
只会设置宿主机上 compose 客户端进程的环境变量，传不进容器。）
|
||
|
||
逐语句执行 + 逐语句报错：一条失败（比如 ALTER 缺权限）不影响其余语句，
|
||
这样你能准确看到是哪一条没过，而不是整个脚本回滚。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
import psycopg
|
||
|
||
import config
|
||
|
||
DEFAULT_SQL = "sql/astock_kg_slot_views.sql"
|
||
|
||
|
||
def split_statements(sql: str) -> list[str]:
|
||
"""按分号切分 SQL 语句。剥掉 -- 行注释，尊重单引号字符串。
|
||
|
||
本文件的 DDL 里没有 $$ 引用块与多行注释，故不做处理；
|
||
若将来加了函数体，这里要先升级再用。
|
||
"""
|
||
out, buf = [], []
|
||
in_str = in_comment = False
|
||
i, n = 0, len(sql)
|
||
while i < n:
|
||
c = sql[i]
|
||
nxt = sql[i + 1] if i + 1 < n else ""
|
||
if in_comment:
|
||
if c == "\n":
|
||
in_comment = False
|
||
buf.append(c)
|
||
i += 1
|
||
continue
|
||
if in_str:
|
||
buf.append(c)
|
||
if c == "'":
|
||
if nxt == "'": # 转义的单引号 ''
|
||
buf.append(nxt)
|
||
i += 2
|
||
continue
|
||
in_str = False
|
||
i += 1
|
||
continue
|
||
if c == "-" and nxt == "-":
|
||
in_comment = True
|
||
i += 2
|
||
continue
|
||
if c == "'":
|
||
in_str = True
|
||
buf.append(c)
|
||
i += 1
|
||
continue
|
||
if c == ";":
|
||
stmt = "".join(buf).strip()
|
||
if stmt:
|
||
out.append(stmt)
|
||
buf = []
|
||
i += 1
|
||
continue
|
||
buf.append(c)
|
||
i += 1
|
||
tail = "".join(buf).strip()
|
||
if tail:
|
||
out.append(tail)
|
||
return out
|
||
|
||
|
||
def _label(stmt: str) -> str:
|
||
"""给语句起个人能读的名字，用于逐条报告。"""
|
||
head = " ".join(stmt.split())[:78]
|
||
return head + ("…" if len(" ".join(stmt.split())) > 78 else "")
|
||
|
||
|
||
def apply(path: str = DEFAULT_SQL, dry_run: bool = False) -> int:
|
||
p = Path(path)
|
||
if not p.exists():
|
||
p = Path("/app") / path # 容器内工作目录兜底
|
||
if not p.exists():
|
||
raise SystemExit(f"找不到 SQL 文件: {path}")
|
||
stmts = split_statements(p.read_text(encoding="utf-8"))
|
||
c = config.akg_pg()
|
||
print(f"应用 {p}（{len(stmts)} 条语句）→ {c.user}@{c.host}:{c.port}/{c.db}")
|
||
if dry_run:
|
||
for i, s in enumerate(stmts, 1):
|
||
print(f" [{i:>2}] {_label(s)}")
|
||
print("\n(dry-run，未执行)")
|
||
return 0
|
||
|
||
ok = fail = 0
|
||
# autocommit：逐条独立提交，一条失败不毒化后续（DDL 各自独立、且都幂等）
|
||
with psycopg.connect(host=c.host, port=c.port, user=c.user,
|
||
password=c.password, dbname=c.db, autocommit=True) as conn:
|
||
for i, s in enumerate(stmts, 1):
|
||
try:
|
||
conn.execute(s)
|
||
ok += 1
|
||
print(f" ✅ [{i:>2}] {_label(s)}")
|
||
except Exception as e: # noqa: BLE001 —— 逐条报错才看得出是哪条缺权限
|
||
fail += 1
|
||
print(f" ❌ [{i:>2}] {_label(s)}\n {e!r}")
|
||
print(f"\n完成：成功 {ok} 条，失败 {fail} 条")
|
||
if fail:
|
||
print("提示：CREATE VIEW / ALTER TABLE 需要 DDL 权限；只读账号会报 "
|
||
"permission denied —— 用基座 owner 账号跑一次即可（见本文件 docstring）。")
|
||
return fail
|