使用带有 sqlalchemy 的 postgresql xml 数据类型

Posted 2023-03-31

【中文标题】使用带有 sqlalchemy 的 postgresql xml 数据类型

【英文标题】：Using postgresql xml data type with sqlalchemy

【发布时间】：2013-04-22 17:50:47

【问题描述】：

SqlAlchemy 通过方言支持大多数特定于数据库的数据类型，但我找不到任何可以使用 postgresql xml 列类型的东西。有人知道可行的解决方案吗？理想情况下，它不应该需要我自己实现自定义列类型。

【问题讨论】：

我没有看到它在 sqlalchemy 0.8 的 postgresql 方言中实现。您可以尝试使用反射来加载具有 XML 列的现有 psql 模式，并查看 sqlalchemy 如何优雅地处理它。但即使在最好的情况下，我也不认为你会得到像 XPATH 索引这样的好东西，即使反射确实做了一些优雅的事情，比如将列公开为文本类型。尝试在 sqlalchemy 问题跟踪器中提交增强请求。

还可以看看 zzzeek 使用邻接表存储 XML 数据的示例：bitbucket.org/sqlalchemy/sqlalchemy/src/…

对于这些类型，我们尽量依赖 psycopg2，虽然它们支持 JSON 和 HSTORE，但它们没有 XML 类型：initd.org/psycopg/docs/extras.html#additional-data-types。也就是说，您仍然可以使用docs.sqlalchemy.org/en/rel_0_8/core/types.html#types-custom 的技术在这里获得很多功能，我们当然会接受完整代码的拉取请求。

【参考方案1】：

如果你需要在 postgresql 数据库中有 native 'xml' 数据类型，你需要编写继承自 UserDefinedType 而非 TypeDecorator 的自定义类型。 Documentation

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

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

【讨论】：

在python 3中，将etree.tostring(value)替换为etree.tostring(value, encoding="unicode")

【参考方案2】：

见：SQLAlchemy TypeDecorator doesn't work

这是修改后的相同解决方案，用于处理具有任意 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)