66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""检查 FDW 前提条件：角色是否存在、app schema 是否有视图/表"""
|
||
import psycopg2
|
||
|
||
CONN = dict(host="100.64.0.4", port=5432, user="local-Python", password="Neo-local-1991125")
|
||
|
||
print("=== 1. 检查角色 ===")
|
||
c = psycopg2.connect(**CONN, dbname="postgres")
|
||
cur = c.cursor()
|
||
for role in ["app_reader", "app_user"]:
|
||
cur.execute("SELECT 1 FROM pg_roles WHERE rolname = %s", (role,))
|
||
exists = cur.fetchone() is not None
|
||
print(f" {role}: {'OK' if exists else 'MISSING'}")
|
||
c.close()
|
||
|
||
print("\n=== 2. 检查 etl_feiqiu.app schema 对象 ===")
|
||
c = psycopg2.connect(**CONN, dbname="etl_feiqiu")
|
||
cur = c.cursor()
|
||
cur.execute(
|
||
"SELECT tablename FROM pg_tables WHERE schemaname = 'app' ORDER BY tablename"
|
||
)
|
||
tables = [r[0] for r in cur.fetchall()]
|
||
cur.execute(
|
||
"SELECT viewname FROM pg_views WHERE schemaname = 'app' ORDER BY viewname"
|
||
)
|
||
views = [r[0] for r in cur.fetchall()]
|
||
cur.execute(
|
||
"SELECT matviewname FROM pg_matviews WHERE schemaname = 'app' ORDER BY matviewname"
|
||
)
|
||
mvs = [r[0] for r in cur.fetchall()]
|
||
print(f" tables: {len(tables)} {tables[:5]}{'...' if len(tables)>5 else ''}")
|
||
print(f" views: {len(views)} {views[:5]}{'...' if len(views)>5 else ''}")
|
||
print(f" matviews: {len(mvs)} {mvs[:5]}{'...' if len(mvs)>5 else ''}")
|
||
c.close()
|
||
|
||
print("\n=== 3. 检查 test_etl_feiqiu.app schema 对象 ===")
|
||
c = psycopg2.connect(**CONN, dbname="test_etl_feiqiu")
|
||
cur = c.cursor()
|
||
cur.execute(
|
||
"SELECT tablename FROM pg_tables WHERE schemaname = 'app' ORDER BY tablename"
|
||
)
|
||
tables2 = [r[0] for r in cur.fetchall()]
|
||
cur.execute(
|
||
"SELECT viewname FROM pg_views WHERE schemaname = 'app' ORDER BY viewname"
|
||
)
|
||
views2 = [r[0] for r in cur.fetchall()]
|
||
cur.execute(
|
||
"SELECT matviewname FROM pg_matviews WHERE schemaname = 'app' ORDER BY matviewname"
|
||
)
|
||
mvs2 = [r[0] for r in cur.fetchall()]
|
||
print(f" tables: {len(tables2)} {tables2[:5]}{'...' if len(tables2)>5 else ''}")
|
||
print(f" views: {len(views2)} {views2[:5]}{'...' if len(views2)>5 else ''}")
|
||
print(f" matviews: {len(mvs2)} {mvs2[:5]}{'...' if len(mvs2)>5 else ''}")
|
||
c.close()
|
||
|
||
print("\n=== 4. 检查 postgres_fdw 扩展可用性 ===")
|
||
for db in ["zqyy_app", "test_zqyy_app"]:
|
||
c = psycopg2.connect(**CONN, dbname=db)
|
||
cur = c.cursor()
|
||
cur.execute("SELECT 1 FROM pg_available_extensions WHERE name = 'postgres_fdw'")
|
||
avail = cur.fetchone() is not None
|
||
cur.execute("SELECT 1 FROM pg_extension WHERE extname = 'postgres_fdw'")
|
||
installed = cur.fetchone() is not None
|
||
print(f" {db}: available={avail}, installed={installed}")
|
||
c.close()
|