SQLAlchemy自定义Oracle Geometry类型遇大几何数据绑定问题求助

因为Geoalchemy2不支持Oracle，我在SQLAlchemy中自定义了OracleGeometry类型以实现基础几何数据支持。小体量几何数据可正常运行，但处理含大量坐标的大几何数据时，WKB/WKT格式数据超出VARCHAR2长度限制，触发错误：

(oracledb.exceptions.DatabaseError) ORA-01461: The value at bind position 2 exceeded the maximum VARCHAR2 length.

我尝试强制使用LOB作为输入但未成功，转而尝试构造Oracle原生SDO_GEOMETRY结构，却遭遇错误：

sqlalchemy.exc.NotSupportedError: (oracledb.exceptions.NotSupportedError) DPY-3002: Python value of type "list" is not supported。

使用环境 #

• Python 3.12
• SQLAlchemy 2.0.29
• oracledb 2.0.1
• shapely 2.0.2

初始自定义类型代码 #

from shapely import from_wkt, Point, wkb
from sqlalchemy import types, Integer, String, create_engine, Column, func
from sqlalchemy.orm import declarative_base, Session
Base = declarative_base()
metadata = Base.metadata


class OracleGeometry(types.UserDefinedType):
    cache_ok = True


    def __init__(self, srid=4326):
        self.srid = srid

    def get_col_spec(self, **kw):
        return "SDO_GEOMETRY"

    def bind_expression(self, bindvalue: Point):
        # adding the nvl2 to make sure it doesn't crash if a null value is added
        return func.nvl2(bindvalue, func.sdo_util.FROM_WKBGEOMETRY(bindvalue), None)

    def column_expression(self, col):
        return func.sdo_util.TO_WKTGEOMETRY(col, type_=self)

    def bind_processor(self, dialect):
        def process(value):
            if value is None:
                return None
            return value.wkb_hex
        return process


    def result_processor(self, dialect, coltype):
        def process(value):
            if value is None:
                return None
            return from_wkt(value)

        return process


class Test2(Base):
    __tablename__ = 'test2'

    id = Column(Integer, primary_key=True)
    test = Column(OracleGeometry(4326))


engine = create_engine(
    f"oracle+oracledb://belmap:belmap@localhost:1521?service_name=FREE", echo=True)

# Base.metadata.create_all(engine)
with Session(engine) as session:
    session.query(Test2).delete()
    wkt = 'MULTIPOLYGON(((....' # Geometry exceeding 4000 characters in wkb or wkt
    t = Test2(id=123, test=from_wkt(wkt))
    t2 = Test2(id=12345, test=None) # Always testing to see if null values work
    session.add_all([t])
    for x in session.query(Test2).all():
        print(f"{x.id} {x.test}")
    session.commit()

改写后的SDO_GEOMETRY构造代码 #

from typing import Optional, List
from shapely.geometry import Polygon
from shapely import from_wkt
from sqlalchemy import types, func

class OracleGeometry(types.UserDefinedType):
    cache_ok = True

    def __init__(self, srid=4326):
        self.srid = srid

    def get_col_spec(self, **kw):
        return "SDO_GEOMETRY"

    def bind_expression(self, bindvalue: Optional[List]):
        # return bindvalue
        return func.nvl2(bindvalue, func.sdo_geometry(bindvalue), None)

    def column_expression(self, col):
        return func.sdo_util.TO_WKTGEOMETRY(col, type_=self)

    def bind_processor(self, dialect):
        def process(geom):
            if geom is None:
                return None
            # return BindParameter(value=value.wkb_hex, type_=BLOB)#str(value.wkb_hex).encode()
            # return value.wkb_hex.encode()
            sdo_gtype = 2007
            sdo_point = None
            sdo_elem_array = []
            sdo_coordinates = []
            for g in geom.geoms:
                g: Polygon
                sdo_elem_array.extend([len(sdo_coordinates) + 1,1003,1])
                for c in g.exterior.coords:
                    sdo_coordinates.extend([c[0], c[1]])
                for i in g.interiors:
                    sdo_elem_array.extend([len(sdo_coordinates) + 1, 2003,1])
                    for c in i.coords:
                        sdo_coordinates.extend([c[0], c[1]])
            return [sdo_gtype, self.srid, sdo_point, sdo_elem_array, sdo_coordinates]
        return process


    def result_processor(self, dialect, coltype):
        def process(value):
            if value is None:
                return None
            return from_wkt(value)

        return process

内容的提问来源于stack exchange，提问作者Lennert De Feyter