Oracle到PostgreSQL地理数据迁移：SDO_Geometry转换失败求助

问题概述 #

现有Python代码可完成Oracle数据库连接、Pandas数据读取、PostgreSQL连接及目标表创建，但数据插入阶段失败，报错：

psycopg2.ProgrammingError: can't adapt type 'DbObject'

原因是Oracle的SDO_Geometry类型被Pandas读取后保留为DbObject类型，psycopg2无法将其适配为PostgreSQL兼容的格式。

核心原因 #

Pandas读取Oracle的SDO_Geometry字段时，不会自动转换为WKT/WKB等文本格式，而是保留为Oracle原生的DbObject对象。psycopg2没有内置适配器处理该类型，导致插入PostgreSQL时触发适配错误。

解决方案 #

最直接高效的方式是在Oracle查询阶段就将SDO_Geometry转换为WKT格式，利用Oracle内置函数SDO_UTIL.TO_WKTGEOMETRY，这样Pandas读取到的是字符串类型，psycopg2可直接处理，插入PostgreSQL时结合PostGIS的ST_GeomFromText函数转换为GEOMETRY类型。

修改步骤 #

1. 调整Oracle数据查询语句，动态识别SDO_Geometry列并转换为WKT格式，其他列保持原样。
2. 修改PostgreSQL插入逻辑，对WKT格式的几何字符串使用ST_GeomFromText转换后插入GEOMETRY列。
3. 移除代码中未导入的wkt/wkb模块相关无效逻辑。

修改后的完整代码 #

import pandas as pd
import psycopg2
from psycopg2 import sql
from psycopg2 import OperationalError, ProgrammingError, IntegrityError
import oracledb

oracledb.init_oracle_client(lib_dir=r"E:\Oracle\product\19.3.0\client_x64\bin")

# 连接Oracle数据库
try:
    oracle_conn = oracledb.connect(user='', password='', host='ws2', port=1521, service_name='pgis.stadt.winroot.net')  
    print("Oracle connection established.")
except oracledb.DatabaseError as e:
    print(f"Error connecting to Oracle: {e}")
    raise

# 查询表结构
schema_name = "WT_RP"
table_name = "W_RP_V_DSP_RP_REGIONAL_L"
query = f"""
SELECT column_name, data_type 
FROM all_tab_columns 
WHERE table_name = '{table_name.upper()}' 
AND owner = '{schema_name.upper()}'
"""
try:
    columns_df = pd.read_sql(query, con=oracle_conn)
    if columns_df.empty:
        print("No columns found for the specified table.")
    else:
        print("Column information retrieved from Oracle.")
except Exception as e:
    print(f"Error retrieving column information from Oracle: {e}")
    oracle_conn.close()
    raise

# 清理列名
columns_df.columns = columns_df.columns.str.strip().str.lower()

# 构建带几何转换的数据查询语句
select_clauses = []
geometry_column = None
for idx, row in columns_df.iterrows():
    col_name = row['column_name']
    data_type = row['data_type'].split('(')[0].lower()
    if data_type == 'sdo_geometry':
        select_clauses.append(f"SDO_UTIL.TO_WKTGEOMETRY({col_name}) AS {col_name}")
        geometry_column = col_name
    else:
        select_clauses.append(col_name)

data_query = f"SELECT {', '.join(select_clauses)} FROM {table_name}"

# 读取数据
try:
    data_df = pd.read_sql(data_query, con=oracle_conn)
    print("Data retrieved from Oracle.")
except Exception as e:
    print(f"Error retrieving data from Oracle: {e}")
    oracle_conn.close()
    raise
finally:
    oracle_conn.close()
    print("Oracle connection closed.")

# Oracle到PostgreSQL数据类型映射
def map_datatypes(oracle_type):
    mapping = {
        "varchar2": "VARCHAR",
        "nvarchar2": "VARCHAR",
        "char": "CHAR",
        "nchar": "CHAR",
        "clob": "TEXT",
        "nclob": "TEXT",
        "blob": "BYTEA",
        "number": "NUMERIC",
        "float": "FLOAT",
        "binary_float": "REAL",
        "binary_double": "DOUBLE PRECISION",
        "date": "DATE",
        "timestamp": "TIMESTAMP",
        "timestamp with time zone": "TIMESTAMP WITH TIME ZONE",
        "timestamp with local time zone": "TIMESTAMP",
        "interval year to month": "INTERVAL YEAR TO MONTH",
        "interval day to second": "INTERVAL DAY TO SECOND",
        "raw": "BYTEA",
        "long": "TEXT",
        "long raw": "BYTEA",
        "rowid": "VARCHAR",
        "urowid": "VARCHAR",
        "xmltype": "XML",
        "sdo_geometry": "GEOMETRY"
    }
    oracle_type_base = oracle_type.split('(')[0].lower()
    return mapping.get(oracle_type_base, "TEXT")

columns_df['postgres_type'] = columns_df['data_type'].apply(map_datatypes)

# 连接PostgreSQL数据库
try:
    pg_conn = psycopg2.connect(
        host="w",
        database="ge",
        user="b",
        password=""
    )
    pg_cursor = pg_conn.cursor()
    print("PostgreSQL connection established.")
    
except OperationalError as e:
    print(f"Error connecting to PostgreSQL: {e}")
    raise

try:
    # 激活PostGIS扩展
    pg_cursor.execute("CREATE EXTENSION IF NOT EXISTS postgis;")
    pg_conn.commit()
    print("PostGIS extension activated.")

    # 创建PostgreSQL表
    create_table_query = sql.SQL("""
    CREATE TABLE IF NOT EXISTS {table} (
        {fields}
    )
    """).format(
        table=sql.Identifier("test_tabelle3"),
        fields=sql.SQL(', ').join(
            sql.SQL("{} {}").format(sql.Identifier(row['column_name']), sql.SQL(row['postgres_type']))
            for idx, row in columns_df.iterrows()
        )
    )
    pg_cursor.execute(create_table_query)
    pg_conn.commit()
    print("Table created in PostgreSQL.")

except Exception as e:
    print(f"Error during table creation: {e}")
    raise   

try:    
    # 构建插入语句，对几何列使用ST_GeomFromText转换
    columns = []
    placeholders = []
    for col in data_df.columns:
        columns.append(sql.Identifier(col))
        if col == geometry_column and geometry_column is not None:
            placeholders.append(sql.SQL("ST_GeomFromText({})").format(sql.Placeholder()))
        else:
            placeholders.append(sql.Placeholder())

    insert_query = sql.SQL("""
    INSERT INTO test_tabelle3 ({columns})
    VALUES ({values})
    """).format(
        columns=sql.SQL(', ').join(columns),
        values=sql.SQL(', ').join(placeholders)
    )

    # 批量插入数据
    for row in data_df.itertuples(index=False, name=None):
        pg_cursor.execute(insert_query, row)
    
    pg_conn.commit()
    print("Data inserted into PostgreSQL.")
        
except (ProgrammingError, IntegrityError) as e:
    print(f"Error during data insertion: {e}")
    pg_conn.rollback()
    raise
finally:
    pg_cursor.close()
    pg_conn.close()
    print("PostgreSQL connection closed.")  

关键修改说明 #

• Oracle查询阶段转换几何类型：通过SDO_UTIL.TO_WKTGEOMETRY将SDO_Geometry直接转为WKT字符串，避免客户端处理DbObject。
• PostgreSQL插入时转换：使用PostGIS的ST_GeomFromText函数将WKT字符串转为PostgreSQL的GEOMETRY类型，适配目标列类型。
• 移除无效代码：删除了未导入的wkt/wkb模块相关逻辑，简化数据处理流程。

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