使用sqlalchemy的postgresql xml数据类型

Ach*_*him 5 python xml sqlalchemy postgresql-9.2

SqlAlchemy通过方言支持大多数特定于数据库的数据类型,但我找不到任何与postgresql xml列类型一起使用的东西.有人知道一个有效的解决方案.理想情况下,我自己不应该要求自定义列类型.

mna*_*ach 5

如果您需要在 postgresql 数据库中拥有本机“xml”数据类型,则需要编写继承自UserDefinedType而不是 TypeDecorator 的自定义类型。文档

这是我在其中一个项目中使用的:

import xml.etree.ElementTree as etree
import sqlalchemy

class XMLType(sqlalchemy.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
Run Code Online (Sandbox Code Playgroud)


Sky*_*ach 2

请参阅:SQLAlchemy TypeDecorator 不起作用

下面是相同的解决方案,经过修改,可以处理任意长度的 xml 的 oracle 的 XMLTYPE,并允许 lxml etree 分配给类列或从类列分配(不需要从容器类中解析/重新解析 xml)

# coding: utf-8
from sqlalchemy import Column, DateTime, Float, ForeignKey, Index, Numeric, String, Table, Text, CLOB
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.sql.functions import GenericFunction
class XMLTypeFunc(GenericFunction):
    type=CLOB
    name='XMLType'
    identifier='XMLTypeFunc'


from sqlalchemy.types import TypeDecorator
from lxml import etree #you can use built-in etree if you want
class XMLType(TypeDecorator):

    impl = CLOB
    type = 'XMLTYPE' #etree.Element

    def get_col_spec(self):
        return 'XMLTYPE'

    def bind_processor(self, dialect):
        def process(value):
            if value is not None:
                return etree.tostring(value, encoding='UTF-8', pretty_print='True')
                #return etree.dump(value)
            else:
                return None
        return process

    def process_result_value(self, value, dialect):
        if value is not None:
            value = etree.fromstring(value)
        return value

    def bind_expression(self, bindvalue):
        return XMLTypeFunc(bindvalue)
Run Code Online (Sandbox Code Playgroud)