SQLAlchemy 在 MSSQL 中声明自定义 UserDefinedType for XML

Question

我正在尝试声明一种自定义类型的"XMLType"以用于MSSQL。但我不断得到'AttributeError: module 'app.db.XMLType' has no attribute '_set_parent_with_dispatch''.我在 XMLType 类文件中没有看到任何未正确声明的内容。

import sqlalchemy.types as types
from lxml import etree
class XMLType(types.UserDefinedType):
    def get_col_spec(self):
        return 'XML'
    def bind_processor(self, dialect):
        def process(value):
            if value is not None:
                if isinstance(value, str):
                    return value
                else:
                    return etree.tostring(value)
            else:
                return None
        return process
    def result_processor(self, dialect, coltype):
        def process(value):
            if value is not None:
                value = etree.fromstring(value)
            return value
        return process

我的代码基于 http://docs.sqlalchemy.org/en/latest/core/custom_types.html?highlight=get_col_spec#sqlalchemy.types.UserDefinedType 和将postgresql xml数据类型与sqlalchemy一起使用。

Answer 1

所以我的导入碰巧出了问题。对于我的模型，我最初有以下内容：

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
import conductor.db.XMLType as XML
Base = declarative_base()
class Raw_data_records(Base):
    __tablename__ = 'raw_data_records'
    id = Column('id', Integer, primary_key=True, autoincrement=True)
    basename = Column('basename', String)
    filename = Column('filename', String)
    file_size = Column('file_size', Integer)
    machine = Column('machine', String)
    insert_timestamp = Column('insert_timestamp', DateTime)
    raw_xml = Column('raw_xml', XML)

然后，我将导入更改为：

from spcconductor.db.XMLType import XMLType和raw_xml = Column('raw_xml', XML) raw_xml = Column('raw_xml', XMLType)，它奏效了。

对于任何需要信息的人来说，这是 XMLType.py(我还没有测试过......

import sqlalchemy.types as types
from lxml import etree
class XMLType(types.UserDefinedType):
    def get_col_spec(self):
        return 'XML'
    def bind_processor(self, dialect):
        def process(value):
            if value is not None:
                if isinstance(value, str):
                    return value
                else:
                    return etree.tostring(value)
            else:
                return None
        return process
    def result_processor(self, dialect, coltype):
        def process(value):
            if value is not None:
                value = etree.fromstring(value)
            return value
        return process