48 lines
1.2 KiB
Python
48 lines
1.2 KiB
Python
from psycopg2 import Error as Psycopg2Error
|
|
|
|
def execute_query(cur, query):
|
|
try:
|
|
# 使用传入的 cursor 执行 SQL 查询
|
|
cur.execute(query)
|
|
# 获取查询结果
|
|
results = cur.fetchall()
|
|
return results
|
|
except Psycopg2Error as e:
|
|
# 这里可以打印异常信息，方便调试
|
|
print(e)
|
|
return "error"
|
|
|
|
|
|
# 获取表名
|
|
def get_table_name(cur):
|
|
# 执行 SQL 查询以获取所有表的列表
|
|
cur.execute("""
|
|
SELECT table_name
|
|
FROM information_schema.tables
|
|
WHERE table_schema = 'public';
|
|
""")
|
|
|
|
# 获取查询结果
|
|
tables = cur.fetchall()
|
|
tname = []
|
|
# 打印表名
|
|
for table in tables:
|
|
tname.append(table[0])
|
|
|
|
return tname
|
|
|
|
# 获取数据库列名。
|
|
def get_table_columns(cur,table_name):
|
|
try:
|
|
# 执行 SQL 查询以获取表的列名
|
|
cur.execute(f"""
|
|
SELECT column_name
|
|
FROM information_schema.columns
|
|
WHERE table_name = '{table_name}';
|
|
""")
|
|
# 获取查询结果
|
|
columns = [desc[0] for desc in cur.fetchall()]
|
|
|
|
except Exception as e:
|
|
print(f"An error occurred: {e}")
|
|
return columns |