如何使用SQLAlchemy向Oracle数据库插入自定义对象类型数据

错误原因 #

你之前的bind_processor返回的是字符串形式的对象构造语句，SQLAlchemy会将这个字符串当作CHAR类型的值传递给Oracle，但Oracle期望接收的是ADDRESS_TYPE类型的对象，因此触发ORA-00932类型不匹配错误。

方案一：利用Oracle驱动原生对象支持（推荐） #

直接通过底层Oracle驱动（如oracledb/cx_Oracle）创建自定义类型的实例，让数据库直接识别为目标类型：

from sqlalchemy import create_engine, Column, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.types import UserDefinedType

Base = declarative_base()

class AddressType(UserDefinedType):
    def get_col_spec(self):
        return "ADDRESS_TYPE"

    def bind_processor(self, dialect):
        def process(value):
            if value is None:
                return None
            # 获取数据库中自定义类型的定义
            raw_conn = dialect.raw_connection()
            address_otype = raw_conn.gettype("ADDRESS_TYPE")
            # 创建自定义类型实例并赋值
            address_obj = address_otype.newobject()
            address_obj.STREET = value['street']
            address_obj.CITY = value['city']
            address_obj.STATE = value['state']
            address_obj.POSTAL_CODE = value['postal_code']
            return address_obj
        return process

    def result_processor(self, dialect, coltype):
        def process(value):
            if value is not None:
                return {
                    'street': value.STREET,
                    'city': value.CITY,
                    'state': value.STATE,
                    'postal_code': value.POSTAL_CODE
                }
            return None
        return process

class MyEntity(Base):
    __tablename__ = 'MY_TABLE'

    id = Column(Integer, primary_key=True)
    custom_address = Column(AddressType, nullable=False)

# 插入逻辑保持不变
new_entity = MyEntity(id=1, custom_address={'street':'street', 'city':'city', 'state':'ST', 'postal_code':'000000'})
db_session.add(new_entity)
db_session.commit()

方案二：使用参数化SQL表达式构造对象 #

通过bind_expression生成参数化的对象构造语句，避免字符串拼接带来的类型错误和SQL注入风险：

from sqlalchemy import create_engine, Column, Integer, text, bindparam
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.types import UserDefinedType

Base = declarative_base()

class AddressType(UserDefinedType):
    def get_col_spec(self):
        return "ADDRESS_TYPE"

    def bind_expression(self, bindvalue):
        # 构造参数化的对象创建SQL
        return text(
            "ADDRESS_TYPE(:street, :city, :state, :postal_code)"
        ).bindparams(
            bindparam("street", bindvalue['street']),
            bindparam("city", bindvalue['city']),
            bindparam("state", bindvalue['state']),
            bindparam("postal_code", bindvalue['postal_code'])
        )

    def result_processor(self, dialect, coltype):
        def process(value):
            if value is not None:
                return {
                    'street': value.STREET,
                    'city': value.CITY,
                    'state': value.STATE,
                    'postal_code': value.POSTAL_CODE
                }
            return None
        return process

class MyEntity(Base):
    __tablename__ = 'MY_TABLE'

    id = Column(Integer, primary_key=True)
    custom_address = Column(AddressType, nullable=False)

# 插入逻辑保持不变
new_entity = MyEntity(id=1, custom_address={'street':'street', 'city':'city', 'state':'ST', 'postal_code':'000000'})
db_session.add(new_entity)
db_session.commit()

方案对比 #

• 方案一：依赖Oracle驱动的原生API，类型匹配精准，性能更优，完全规避SQL注入风险。
• 方案二：纯SQLAlchemy层面实现，不依赖底层驱动细节，通用性更强，同样通过参数化查询避免安全问题。

注意事项 #

确保数据库中自定义类型名称（ADDRESS_TYPE）与get_col_spec返回值一致，Oracle默认不区分大小写，但如果创建类型时使用了引号包裹，需严格匹配大小写。

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