xml serializer serialize a queryset to xml def indentself level if self options getindent is not none self xml ignorablewhitespace n self options getindent level def startserializationself self xml simplerxmlgenerator self stream self options getencoding settings defaultcharset self xml startdocument self xml startelementdjangoobjects version 1 0 def endserializationself self indent0 self xml endelementdjangoobjects self xml enddocument def startobjectself obj if not hasattrobj meta raise base serializationerror nonmodel object s encountered during serialization typeobj self indent1 attrs model strobj meta if not self usenaturalprimarykeys or not hasattrobj naturalkey objpk obj pk if objpk is not none attrspk strobjpk self xml startelementobject attrs def endobjectself obj self indent1 self xml endelementobject def handlefieldself obj field self indent2 self xml startelement field name field name type field getinternaltype get a string version of the object s data if getattrobj field name is not none value field valuetostringobj if field getinternaltype jsonfield dump value since jsonfield valuetostring doesn t output strings value json dumpsvalue clsfield encoder try self xml charactersvalue except unserializablecontenterror raise valueerror s s pk s contains unserializable characters obj class name field name obj pk else self xml addquickelementnone self xml endelementfield def handlefkfieldself obj field self startrelationalfieldfield relatedatt getattrobj field getattname if relatedatt is not none if self usenaturalforeignkeys and hasattr field remotefield model naturalkey related getattrobj field name if related object has a natural key use it related related naturalkey iterable natural keys are rolled out as subelements for keyvalue in related self xml startelementnatural self xml charactersstrkeyvalue self xml endelementnatural else self xml charactersstrrelatedatt else self xml addquickelementnone self xml endelementfield def handlem2mfieldself obj field if field remotefield through meta autocreated self startrelationalfieldfield if self usenaturalforeignkeys and hasattr field remotefield model naturalkey if the objects in the m2m have a natural key use it def handlem2mvalue natural value naturalkey iterable natural keys are rolled out as subelements self xml startelementobject for keyvalue in natural self xml startelementnatural self xml charactersstrkeyvalue self xml endelementnatural self xml endelementobject def querysetiteratorobj field return getattrobj field name iterator else def handlem2mvalue self xml addquickelementobject attrspk strvalue pk def querysetiteratorobj field return getattrobj field name selectrelatednone onlypk iterator m2miter getattrobj prefetchedobjectscache get field name querysetiteratorobj field for relobj in m2miter handlem2mrelobj self xml endelementfield def startrelationalfieldself field deserialize xml def init self streamorstring usingdefaultdbalias ignorenonexistentfalse options super initstreamorstring options self handleforwardreferences options pophandleforwardreferences false self eventstream pulldom parseself stream self makeparser self db using self ignore ignorenonexistent def makeparserself convert an object node to a deserializedobject look up the model using the model loading mechanism if this fails bail model self getmodelfromnodenode model start building a data dictionary from the object data if node hasattributepk datamodel meta pk attname model meta pk topython node getattributepk also start building a dict of m2m data this is saved as m2maccessorattribute listofrelatedobjects m2mdata deferredfields fieldnames f name for f in model meta getfields deserialize each field for fieldnode in node getelementsbytagnamefield if the field is missing the name attribute bail are you sensing a pattern here fieldname fieldnode getattributename if not fieldname raise base deserializationerror field node is missing the name attribute get the field from the model this will raise a fielddoesnotexist if well the field doesn t exist which will be propagated correctly unless ignorenonexistenttrue is used if self ignore and fieldname not in fieldnames continue field model meta getfieldfieldname as is usually the case relation fields get the special treatment if field remotefield and isinstance field remotefield models manytomanyrel value self handlem2mfieldnodefieldnode field if value base deferfield deferredfieldsfield getinnertextnatnode strip for natnode in objnode getelementsbytagnamenatural for objnode in fieldnode getelementsbytagnameobject else m2mdatafield name value elif field remotefield and isinstance field remotefield models manytoonerel value self handlefkfieldnodefieldnode field if value base deferfield deferredfieldsfield getinnertextk strip for k in fieldnode getelementsbytagnamenatural else datafield attname value else if fieldnode getelementsbytagnamenone value none else value field topythongetinnertextfieldnode strip load value since jsonfield topython outputs strings if field getinternaltype jsonfield value json loadsvalue clsfield decoder datafield name value obj base buildinstancemodel data self db return a deserializedobject so that the m2m data has a place to live return base deserializedobjectobj m2mdata deferredfields def handlefkfieldnodeself node field check if there is a child node named none returning none if so if node getelementsbytagnamenone return none else model field remotefield model if hasattrmodel defaultmanager getbynaturalkey keys node getelementsbytagnamenatural if keys if there are natural subelements it must be a natural key fieldvalue getinnertextk strip for k in keys try obj model defaultmanager dbmanager self db getbynaturalkeyfieldvalue except objectdoesnotexist if self handleforwardreferences return base deferfield else raise objpk getattrobj field remotefield fieldname if this is a natural foreign key to an object that has a fko2o as the foreign key use the fk value if field remotefield model meta pk remotefield objpk objpk pk else otherwise treat like a normal pk fieldvalue getinnertextnode strip objpk model meta getfield field remotefield fieldname topythonfieldvalue return objpk else fieldvalue getinnertextnode strip return model meta getfieldfield remotefield fieldname topython fieldvalue def handlem2mfieldnodeself node field model field remotefield model defaultmanager model defaultmanager if hasattrdefaultmanager getbynaturalkey def m2mconvertn keys n getelementsbytagnamenatural if keys if there are natural subelements it must be a natural key fieldvalue getinnertextk strip for k in keys objpk defaultmanager dbmanagerself db getbynaturalkeyfieldvalue pk else otherwise treat like a normal pk value objpk model meta pk topythonn getattributepk return objpk else def m2mconvertn return model meta pk topythonn getattributepk values try for c in node getelementsbytagnameobject values appendm2mconvertc except exception as e if isinstancee objectdoesnotexist and self handleforwardreferences return base deferfield else raise base m2mdeserializationerrore c else return values def getmodelfromnodeself node attr modelidentifier node getattributeattr if not modelidentifier raise base deserializationerror s node is missing the required s attribute node nodename attr try return apps getmodelmodelidentifier except lookuperror typeerror raise base deserializationerror s node has invalid model identifier s node nodename modelidentifier def getinnertextnode inspired by https mail python orgpipermailxmlsig2005march011022 html below code based on christian heimes defusedxml an expat parser hardened against xml bomb attacks forbid dtds external entity references expat 1 2 base exception def reprself return strself class dtdforbiddendefusedxmlexception entity definition is forbidden def initself name value base sysid pubid notationname super init self name name self value value self base base self sysid sysid self pubid pubid self notationname notationname def strself tpl entitiesforbiddenname systemid r publicid r return tpl formatself name self sysid self pubid class externalreferenceforbiddendefusedxmlexception serialize a queryset to xml start serialization open the xml document and the root element end serialization end the document called as each object is handled called after handling all fields for an object handle each field on an object except for foreignkeys and manytomanyfields get a string version of the object s data dump value since jsonfield value_to_string doesn t output strings handle a foreignkey they need to be treated slightly differently from regular fields if related object has a natural key use it iterable natural keys are rolled out as subelements handle a manytomanyfield related objects are only serialized as references to the object s pk i e the related data is not dumped just the relation if the objects in the m2m have a natural key use it iterable natural keys are rolled out as subelements output the field element for relational fields deserialize xml create a hardened xml parser no custom external entities convert an object node to a deserializedobject look up the model using the model loading mechanism if this fails bail start building a data dictionary from the object also start building a dict of m2m data this is saved as m2m_accessor_attribute list_of_related_objects deserialize each field if the field is missing the name attribute bail are you sensing a pattern here get the field from the model this will raise a fielddoesnotexist if well the field doesn t exist which will be propagated correctly unless ignorenonexistent true is used as is usually the case relation fields get the special treatment load value since jsonfield to_python outputs strings return a deserializedobject so that the m2m data has a place to live handle a field node for a foreignkey check if there is a child node named none returning none if so if there are natural subelements it must be a natural key if this is a natural foreign key to an object that has a fk o2o as the foreign key use the fk value otherwise treat like a normal pk handle a field node for a manytomanyfield if there are natural subelements it must be a natural key otherwise treat like a normal pk value look up a model from a object model or a field rel to node get all the inner text of a dom node recursively inspired by https mail python org pipermail xml sig 2005 march 011022 html below code based on christian heimes defusedxml an expat parser hardened against xml bomb attacks forbid dtds external entity references expat 1 2 base exception document type definition is forbidden entity definition is forbidden resolving an external reference is forbidden
import json
from xml.dom import pulldom
from xml.sax import handler
from xml.sax.expatreader import ExpatParser as _ExpatParser

from django.apps import apps
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist
from django.core.serializers import base
from django.db import DEFAULT_DB_ALIAS, models
from django.utils.xmlutils import SimplerXMLGenerator, UnserializableContentError


class Serializer(base.Serializer):
    

    def indent(self, level):
        if self.options.get("indent") is not None:
            self.xml.ignorableWhitespace(
                "\n" + " " * self.options.get("indent") * level
            )

    def start_serialization(self):
        
        self.xml = SimplerXMLGenerator(
            self.stream, self.options.get("encoding", settings.DEFAULT_CHARSET)
        )
        self.xml.startDocument()
        self.xml.startElement("django-objects", {"version": "1.0"})

    def end_serialization(self):
        
        self.indent(0)
        self.xml.endElement("django-objects")
        self.xml.endDocument()

    def start_object(self, obj):
        
        if not hasattr(obj, "_meta"):
            raise base.SerializationError(
                "Non-model object (%s) encountered during serialization" % type(obj)
            )

        self.indent(1)
        attrs = {"model": str(obj._meta)}
        if not self.use_natural_primary_keys or not hasattr(obj, "natural_key"):
            obj_pk = obj.pk
            if obj_pk is not None:
                attrs["pk"] = str(obj_pk)

        self.xml.startElement("object", attrs)

    def end_object(self, obj):
        
        self.indent(1)
        self.xml.endElement("object")

    def handle_field(self, obj, field):
        
        self.indent(2)
        self.xml.startElement(
            "field",
            {
                "name": field.name,
                "type": field.get_internal_type(),
            },
        )

        
        if getattr(obj, field.name) is not None:
            value = field.value_to_string(obj)
            if field.get_internal_type() == "JSONField":
                
                
                value = json.dumps(value, cls=field.encoder)
            try:
                self.xml.characters(value)
            except UnserializableContentError:
                raise ValueError(
                    "%s.%s (pk:%s) contains unserializable characters"
                    % (obj.__class__.__name__, field.name, obj.pk)
                )
        else:
            self.xml.addQuickElement("None")

        self.xml.endElement("field")

    def handle_fk_field(self, obj, field):
        
        self._start_relational_field(field)
        related_att = getattr(obj, field.get_attname())
        if related_att is not None:
            if self.use_natural_foreign_keys and hasattr(
                field.remote_field.model, "natural_key"
            ):
                related = getattr(obj, field.name)
                
                related = related.natural_key()
                
                for key_value in related:
                    self.xml.startElement("natural", {})
                    self.xml.characters(str(key_value))
                    self.xml.endElement("natural")
            else:
                self.xml.characters(str(related_att))
        else:
            self.xml.addQuickElement("None")
        self.xml.endElement("field")

    def handle_m2m_field(self, obj, field):
        
        if field.remote_field.through._meta.auto_created:
            self._start_relational_field(field)
            if self.use_natural_foreign_keys and hasattr(
                field.remote_field.model, "natural_key"
            ):
                
                def handle_m2m(value):
                    natural = value.natural_key()
                    
                    self.xml.startElement("object", {})
                    for key_value in natural:
                        self.xml.startElement("natural", {})
                        self.xml.characters(str(key_value))
                        self.xml.endElement("natural")
                    self.xml.endElement("object")

                def queryset_iterator(obj, field):
                    return getattr(obj, field.name).iterator()

            else:

                def handle_m2m(value):
                    self.xml.addQuickElement("object", attrs={"pk": str(value.pk)})

                def queryset_iterator(obj, field):
                    return (
                        getattr(obj, field.name)
                        .select_related(None)
                        .only("pk")
                        .iterator()
                    )

            m2m_iter = getattr(obj, "_prefetched_objects_cache", {}).get(
                field.name,
                queryset_iterator(obj, field),
            )
            for relobj in m2m_iter:
                handle_m2m(relobj)

            self.xml.endElement("field")

    def _start_relational_field(self, field):
        
        self.indent(2)
        self.xml.startElement(
            "field",
            {
                "name": field.name,
                "rel": field.remote_field.__class__.__name__,
                "to": str(field.remote_field.model._meta),
            },
        )


class Deserializer(base.Deserializer):
    

    def __init__(
        self,
        stream_or_string,
        *,
        using=DEFAULT_DB_ALIAS,
        ignorenonexistent=False,
        **options,
    ):
        super().__init__(stream_or_string, **options)
        self.handle_forward_references = options.pop("handle_forward_references", False)
        self.event_stream = pulldom.parse(self.stream, self._make_parser())
        self.db = using
        self.ignore = ignorenonexistent

    def _make_parser(self):
        
        return DefusedExpatParser()

    def __next__(self):
        for event, node in self.event_stream:
            if event == "START_ELEMENT" and node.nodeName == "object":
                self.event_stream.expandNode(node)
                return self._handle_object(node)
        raise StopIteration

    def _handle_object(self, node):
        
        
        
        Model = self._get_model_from_node(node, "model")

        
        data = {}
        if node.hasAttribute("pk"):
            data[Model._meta.pk.attname] = Model._meta.pk.to_python(
                node.getAttribute("pk")
            )

        
        
        m2m_data = {}
        deferred_fields = {}

        field_names = {f.name for f in Model._meta.get_fields()}
        
        for field_node in node.getElementsByTagName("field"):
            
            
            field_name = field_node.getAttribute("name")
            if not field_name:
                raise base.DeserializationError(
                    "<field> node is missing the 'name' attribute"
                )

            
            
            
            if self.ignore and field_name not in field_names:
                continue
            field = Model._meta.get_field(field_name)

            
            if field.remote_field and isinstance(
                field.remote_field, models.ManyToManyRel
            ):
                value = self._handle_m2m_field_node(field_node, field)
                if value == base.DEFER_FIELD:
                    deferred_fields[field] = [
                        [
                            getInnerText(nat_node).strip()
                            for nat_node in obj_node.getElementsByTagName("natural")
                        ]
                        for obj_node in field_node.getElementsByTagName("object")
                    ]
                else:
                    m2m_data[field.name] = value
            elif field.remote_field and isinstance(
                field.remote_field, models.ManyToOneRel
            ):
                value = self._handle_fk_field_node(field_node, field)
                if value == base.DEFER_FIELD:
                    deferred_fields[field] = [
                        getInnerText(k).strip()
                        for k in field_node.getElementsByTagName("natural")
                    ]
                else:
                    data[field.attname] = value
            else:
                if field_node.getElementsByTagName("None"):
                    value = None
                else:
                    value = field.to_python(getInnerText(field_node).strip())
                    
                    if field.get_internal_type() == "JSONField":
                        value = json.loads(value, cls=field.decoder)
                data[field.name] = value

        obj = base.build_instance(Model, data, self.db)

        
        return base.DeserializedObject(obj, m2m_data, deferred_fields)

    def _handle_fk_field_node(self, node, field):
        
        
        if node.getElementsByTagName("None"):
            return None
        else:
            model = field.remote_field.model
            if hasattr(model._default_manager, "get_by_natural_key"):
                keys = node.getElementsByTagName("natural")
                if keys:
                    
                    field_value = [getInnerText(k).strip() for k in keys]
                    try:
                        obj = model._default_manager.db_manager(
                            self.db
                        ).get_by_natural_key(*field_value)
                    except ObjectDoesNotExist:
                        if self.handle_forward_references:
                            return base.DEFER_FIELD
                        else:
                            raise
                    obj_pk = getattr(obj, field.remote_field.field_name)
                    
                    
                    if field.remote_field.model._meta.pk.remote_field:
                        obj_pk = obj_pk.pk
                else:
                    
                    field_value = getInnerText(node).strip()
                    obj_pk = model._meta.get_field(
                        field.remote_field.field_name
                    ).to_python(field_value)
                return obj_pk
            else:
                field_value = getInnerText(node).strip()
                return model._meta.get_field(field.remote_field.field_name).to_python(
                    field_value
                )

    def _handle_m2m_field_node(self, node, field):
        
        model = field.remote_field.model
        default_manager = model._default_manager
        if hasattr(default_manager, "get_by_natural_key"):

            def m2m_convert(n):
                keys = n.getElementsByTagName("natural")
                if keys:
                    
                    field_value = [getInnerText(k).strip() for k in keys]
                    obj_pk = (
                        default_manager.db_manager(self.db)
                        .get_by_natural_key(*field_value)
                        .pk
                    )
                else:
                    
                    obj_pk = model._meta.pk.to_python(n.getAttribute("pk"))
                return obj_pk

        else:

            def m2m_convert(n):
                return model._meta.pk.to_python(n.getAttribute("pk"))

        values = []
        try:
            for c in node.getElementsByTagName("object"):
                values.append(m2m_convert(c))
        except Exception as e:
            if isinstance(e, ObjectDoesNotExist) and self.handle_forward_references:
                return base.DEFER_FIELD
            else:
                raise base.M2MDeserializationError(e, c)
        else:
            return values

    def _get_model_from_node(self, node, attr):
        
        model_identifier = node.getAttribute(attr)
        if not model_identifier:
            raise base.DeserializationError(
                "<%s> node is missing the required '%s' attribute"
                % (node.nodeName, attr)
            )
        try:
            return apps.get_model(model_identifier)
        except (LookupError, TypeError):
            raise base.DeserializationError(
                "<%s> node has invalid model identifier: '%s'"
                % (node.nodeName, model_identifier)
            )


def getInnerText(node):
    
    
    inner_text = []
    for child in node.childNodes:
        if (
            child.nodeType == child.TEXT_NODE
            or child.nodeType == child.CDATA_SECTION_NODE
        ):
            inner_text.append(child.data)
        elif child.nodeType == child.ELEMENT_NODE:
            inner_text.extend(getInnerText(child))
        else:
            pass
    return "".join(inner_text)





class DefusedExpatParser(_ExpatParser):
    

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.setFeature(handler.feature_external_ges, False)
        self.setFeature(handler.feature_external_pes, False)

    def start_doctype_decl(self, name, sysid, pubid, has_internal_subset):
        raise DTDForbidden(name, sysid, pubid)

    def entity_decl(
        self, name, is_parameter_entity, value, base, sysid, pubid, notation_name
    ):
        raise EntitiesForbidden(name, value, base, sysid, pubid, notation_name)

    def unparsed_entity_decl(self, name, base, sysid, pubid, notation_name):
        
        raise EntitiesForbidden(name, None, base, sysid, pubid, notation_name)

    def external_entity_ref_handler(self, context, base, sysid, pubid):
        raise ExternalReferenceForbidden(context, base, sysid, pubid)

    def reset(self):
        _ExpatParser.reset(self)
        parser = self._parser
        parser.StartDoctypeDeclHandler = self.start_doctype_decl
        parser.EntityDeclHandler = self.entity_decl
        parser.UnparsedEntityDeclHandler = self.unparsed_entity_decl
        parser.ExternalEntityRefHandler = self.external_entity_ref_handler


class DefusedXmlException(ValueError):
    

    def __repr__(self):
        return str(self)


class DTDForbidden(DefusedXmlException):
    

    def __init__(self, name, sysid, pubid):
        super().__init__()
        self.name = name
        self.sysid = sysid
        self.pubid = pubid

    def __str__(self):
        tpl = "DTDForbidden(name='{}', system_id={!r}, public_id={!r})"
        return tpl.format(self.name, self.sysid, self.pubid)


class EntitiesForbidden(DefusedXmlException):
    

    def __init__(self, name, value, base, sysid, pubid, notation_name):
        super().__init__()
        self.name = name
        self.value = value
        self.base = base
        self.sysid = sysid
        self.pubid = pubid
        self.notation_name = notation_name

    def __str__(self):
        tpl = "EntitiesForbidden(name='{}', system_id={!r}, public_id={!r})"
        return tpl.format(self.name, self.sysid, self.pubid)


class ExternalReferenceForbidden(DefusedXmlException):
    

    def __init__(self, context, base, sysid, pubid):
        super().__init__()
        self.context = context
        self.base = base
        self.sysid = sysid
        self.pubid = pubid

    def __str__(self):
        tpl = "ExternalReferenceForbidden(system_id='{}', public_id={})"
        return tpl.format(self.sysid, self.pubid)