60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
"""查询测试库 DWS 层 cfg_* 配置表的内容"""
|
|
import os
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
import psycopg2
|
|
import psycopg2.extras
|
|
|
|
load_dotenv(Path(__file__).resolve().parents[2] / ".env")
|
|
|
|
dsn = os.environ.get("TEST_DB_DSN")
|
|
if not dsn:
|
|
raise RuntimeError("TEST_DB_DSN 未配置")
|
|
|
|
conn = psycopg2.connect(dsn)
|
|
|
|
with conn.cursor() as cur:
|
|
# 1. 列出所有 cfg_ 表
|
|
cur.execute("""
|
|
SELECT table_name
|
|
FROM information_schema.tables
|
|
WHERE table_schema = 'dws' AND table_name LIKE 'cfg_%%'
|
|
ORDER BY table_name
|
|
""")
|
|
tables = [r[0] for r in cur.fetchall()]
|
|
print(f"=== DWS cfg_* 配置表列表 ({len(tables)} 张) ===")
|
|
for t in tables:
|
|
print(f" - {t}")
|
|
print()
|
|
|
|
# 2. 逐表查询内容
|
|
for t in tables:
|
|
cur.execute(f"SELECT count(*) FROM dws.{t}")
|
|
cnt = cur.fetchone()[0]
|
|
print(f"\n{'='*60}")
|
|
print(f"表: dws.{t} (共 {cnt} 条记录)")
|
|
print('='*60)
|
|
|
|
if cnt == 0:
|
|
print(" (空表)")
|
|
continue
|
|
|
|
cur.execute(f"SELECT * FROM dws.{t} ORDER BY 1")
|
|
cols = [desc[0] for desc in cur.description]
|
|
rows = cur.fetchall()
|
|
|
|
# 打印列头
|
|
print(" " + " | ".join(cols))
|
|
print(" " + "-" * (len(" | ".join(cols)) + 10))
|
|
|
|
for row in rows:
|
|
vals = []
|
|
for v in row:
|
|
if v is None:
|
|
vals.append("NULL")
|
|
else:
|
|
vals.append(str(v))
|
|
print(" " + " | ".join(vals))
|
|
|
|
conn.close()
|