Using postgresql xml data type with sqlalchemy

问题:

SqlAlchemy supports most database specific data types via dialects, but I could not find anything to work with the postgresql xml column type. Does somebody know a working solution. Idealy it should not require a custom column type implementation by myself.

回答1:

If you need to have native 'xml' data type in postgresql database, you need to write custom type which inherited from UserDefinedType not from TypeDecorator. Documentation

Here is what I used in one of the projects:

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

回答2:

See: SQLAlchemy TypeDecorator doesn't work

Here is the same solution modified to handle XMLTYPE for oracle with any length of xml and to allow lxml etree assignment to and from the class column (no need to deparse/reparse xml from container classes)

# 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)