SQL Alchemy 自定义类型，强制绑定 BLOB 参数

Question

我正在尝试为Oracle数据库在SQL Alchemy中添加基本的几何支持，因为Geoalchemy2不支持这一点。我创建了一个自定义数据类型，如果几何图形不太大，它工作得很好。但是，由于我使用的是wkb或wkt作为输入，当几何图形的坐标数量很大时，就会超出varchar2的限制。

(oracledb.exceptions.DatabaseError) ORA-01461: 绑定位置2的值超出了最大VARCHAR2长度。

我想强制它使用LOB作为输入，但似乎无法做到。或者有没有其他简单的方法来解决这个问题……最好是不用麻烦地处理Oracle的几何类型。

这是我目前的代码：

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(

)

顺便说一下，我使用的是：

python 3.12

sqlalchemy 2.0.29

oracledb 2.0.1

shapely 2.0.2

编辑

所以，我决定尝试Oracle的几何定义。现在我的自定义类型看起来像这样（对于多边形）：

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

但现在我遇到了错误：sqlalchemy.exc.NotSupportedError: (oracledb.exceptions.NotSupportedError) DPY-3002: 不支持类型为“list”的Python值。