使用psycopg2创建Schema并设置为默认的问题求助

一、解决Schema名称带单引号的问题 #

PostgreSQL中，Schema名称属于标识符，不能用单引号包裹，而psycopg2的普通参数化查询会自动给字符串值添加单引号，这就是报错的核心原因。正确做法是用psycopg2.sql模块提供的Identifier类安全构造标识符，既避免语法错误，也能防范SQL注入风险。

修正后的创建Schema代码 #

先导入psycopg2.sql模块，再用SQL和Identifier拼接合法的SQL语句：

import psycopg2
from psycopg2.extras import RealDictCursor
from psycopg2 import sql  # 新增导入
import re

# 配置信息保持不变...

# 连接和游标设置保持不变...

key = '36107.csv'
fips = re.sub(".csv", "", key) + "county"

# 用sql模块构造安全的建Schema语句
create_schema_sql = sql.SQL("CREATE SCHEMA IF NOT EXISTS {}").format(sql.Identifier(fips))
cursor.execute(create_schema_sql)

执行后Schema名称会被正确处理：若名称包含特殊字符会自动加双引号，普通名称则直接使用，不会出现多余的单引号。

二、解决search_path不生效的问题 #

你当前用ALTER DATABASE的操作存在两个问题：

1. 生效时机问题：ALTER DATABASE修改的是数据库全局默认search_path，但仅对新创建的连接生效，当前连接不会立即应用该设置，需断开重连才会生效。
2. 数据库名称不匹配：你连接的是postgres数据库，但SQL语句中写的是ALTER DATABASE raster SET...，这会修改raster数据库的设置，和当前会话完全无关。

如果只是想让当前会话的后续操作默认使用目标Schema，推荐使用会话级的SET命令，执行后立即生效，无需重连：

修正后的设置search_path代码 #

# 设置当前会话的search_path，优先使用目标Schema，同时保留public（可选）
set_search_path_sql = sql.SQL("SET search_path TO {}, public").format(sql.Identifier(fips))
cursor.execute(set_search_path_sql)

如果确实需要修改数据库的全局默认search_path（对所有新连接生效），则要确保数据库名称正确，且执行后重新连接：

# 修改数据库全局默认search_path（替换为你实际连接的数据库名）
alter_db_sql = sql.SQL("ALTER DATABASE {} SET search_path TO {}, public").format(
    sql.Identifier(database_db),  # 这里用你连接的数据库名
    sql.Identifier(fips)
)
cursor.execute(alter_db_sql)

# 重新连接才能让全局设置生效
connection.close()
connection = psycopg2.connect(
    dbname=database_db, 
    user=username_db,
    password=password_db,
    host=endpoint_db,
    port=port_db
)
cursor = connection.cursor(cursor_factory=RealDictCursor)

完整修正代码 #

import psycopg2
from psycopg2.extras import RealDictCursor
from psycopg2 import sql
import re

# Configuration Values
endpoint_db = 'localhost'
database_db = 'postgres'
username_db = 'postgres'
password_db = '****'
port_db = 5432

# Connect
connection = psycopg2.connect(
    dbname=database_db, 
    user=username_db,
    password=password_db,
    host=endpoint_db,
    port=port_db
)

# Set the cursor
connection.autocommit = True
cursor = connection.cursor(cursor_factory=RealDictCursor)

# Set the schema name
key = '36107.csv'
fips = re.sub(".csv", "", key) + "county"

# Create schema safely
create_schema_sql = sql.SQL("CREATE SCHEMA IF NOT EXISTS {}").format(sql.Identifier(fips))
cursor.execute(create_schema_sql)

# Set session-level search path (immediately effective)
set_search_path_sql = sql.SQL("SET search_path TO {}, public").format(sql.Identifier(fips))
cursor.execute(set_search_path_sql)

# 后续操作直接写表名即可，默认使用目标Schema
# 示例：cursor.execute("CREATE TABLE test (id INT)") 会创建在fips对应的Schema下

内容的提问来源于stack exchange，提问作者Andrei Niță