Python脚本中PostgreSQL表查询失效及F-string警告问题求助

问题1：F-string触发“F-string is missing placeholders”警告 #

原因 #

这个警告是因为你写的F-string没有包含任何变量插值（即{变量名}格式的占位符），属于完全静态的字符串，没必要用F-string语法。比如这类写法：

# 错误示例：无占位符的冗余F-string
sql = f"SELECT tablename FROM pg_tables WHERE schemaname = 'public' AND tablename NOT LIKE 'pg_%'"

修正方法 #

• 如果SQL是静态内容，直接去掉f前缀即可：

  sql = "SELECT tablename FROM pg_tables WHERE schemaname = 'public' AND tablename NOT LIKE 'pg_%'"
  

• 如果需要动态插入变量（比如指定模式名），不要用F-string拼接SQL（避免SQL注入风险），改用psycopg2的参数化查询：

  schema_name = 'public'
  sql = "SELECT tablename FROM pg_tables WHERE schemaname = %s AND tablename NOT LIKE 'pg_%'"
  cursor.execute(sql, (schema_name,))
  

问题2：SQL在psql生效但Python脚本中无效 #

常见原因及解决方法 #

1. 连接的数据库不一致
   psql默认连接postgres数据库，而你的Python脚本可能连接了其他库。要查询所有数据库的用户表，需遍历数据库列表，逐个连接对应库执行查询：

import psycopg2

# 先连接postgres库获取所有非模板数据库列表
conn = psycopg2.connect(dbname='postgres', user='your_user', password='your_pwd', host='localhost')
conn.autocommit = True
cur = conn.cursor()
cur.execute("SELECT datname FROM pg_database WHERE datistemplate = false")
databases = [row[0] for row in cur.fetchall()]
cur.close()
conn.close()

# 遍历每个数据库查询用户表
for db in databases:
    conn = psycopg2.connect(dbname=db, user='your_user', password='your_pwd', host='localhost')
    cur = conn.cursor()
    cur.execute("SELECT tablename FROM pg_tables WHERE schemaname = 'public' AND tablename NOT LIKE 'pg_%'")
    tables = cur.fetchall()
    print(f"数据库 {db} 的用户表：{tables}")
    cur.close()
    conn.close()

2. 权限差异
   Python脚本使用的数据库用户和你psql登录的用户权限不同，导致无法查看部分表。可以在Python中执行SELECT current_user;查看当前用户，和psql里的用户对比，确保权限一致。

3. SQL格式错误
   如果SQL包含换行或特殊字符，用三重引号包裹避免格式混乱：

sql = """
SELECT tablename 
FROM pg_tables 
WHERE schemaname = 'public' 
  AND tablename NOT LIKE 'pg_%'
"""

4. 未正确获取查询结果
   执行cursor.execute(sql)后，必须调用fetchall()、fetchone()或fetchmany()才能获取结果，否则相当于只执行了查询但没读取返回数据。

5. 跨数据库查询限制
   PostgreSQL不支持在单个连接中直接查询其他数据库的表，必须切换到对应数据库的连接后再执行查询，这也是上面代码中遍历数据库逐个连接的原因。

完整脚本示例（含CSV写入） #

import psycopg2
import csv

def get_postgres_data():
    # 连接postgres库获取基础信息
    conn = psycopg2.connect(dbname='postgres', user='postgres', password='your_password', host='localhost')
    conn.autocommit = True
    cur = conn.cursor()

    # 获取PostgreSQL版本
    cur.execute("SELECT version();")
    pg_version = cur.fetchone()[0]

    # 获取数据库列表及大小
    cur.execute("""
        SELECT 
            datname,
            pg_size_pretty(pg_database_size(datname)) as size
        FROM pg_database 
        WHERE datistemplate = false
    """)
    db_list = cur.fetchall()
    cur.close()
    conn.close()

    # 收集每个数据库的用户表信息
    db_details = []
    for db_name, db_size in db_list:
        conn = psycopg2.connect(dbname=db_name, user='postgres', password='your_password', host='localhost')
        cur = conn.cursor()
        cur.execute("""
            SELECT tablename 
            FROM pg_tables 
            WHERE schemaname = 'public' 
              AND tablename NOT LIKE 'pg_%'
        """)
        tables = [row[0] for row in cur.fetchall()]
        db_details.append({
            'db_name': db_name,
            'db_size': db_size,
            'tables': ', '.join(tables) if tables else '无用户表'
        })
        cur.close()
        conn.close()

    # 写入CSV文件
    with open('postgres_stats.csv', 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=['pg_version', 'db_name', 'db_size', 'tables'])
        writer.writeheader()
        for detail in db_details:
            writer.writerow({
                'pg_version': pg_version,
                'db_name': detail['db_name'],
                'db_size': detail['db_size'],
                'tables': detail['tables']
            })

if __name__ == '__main__':
    get_postgres_data()

内容的提问来源于stack exchange，提问作者Tommy Gibbons