SQL Alchemy custom type, forcing blob bind parameter

Question

I am trying to get basic Geometry support for Oracle into SQL Alchemy since Geoalchemy2 doesn't support it. I've created a custom datatype that works perfectly if geometries aren't too big. But since I use wkb or wkt as an input, large geometries (in terms of nr of coordinates) cause the varchar2 limit to be exceeded.

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

I would like to force it to use a LOB as input, but can't seem to be able to. Or another way to make this work easily... preferably without the headache of dealing with oracle geometry types.

This is the code that I currently have:

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(

)

Fyi, I am using python 3.12 sqlalchemy 2.0.29 oracledb 2.0.1 shapely 2.0.2

EDIT So, I bit the bullet, and am going to try the oracle geometry definitions. Now my custom type looks like (for a multipolygon):

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

But now I get the error: sqlalchemy.exc.NotSupportedError: (oracledb.exceptions.NotSupportedError) DPY-3002: Python value of type "list" is not supported