Stream XML records with iterparse

By Batu · English technical notes

Also published in our Blogger archive.

Stream XML records with iterparse

ElementTree.iterparse() reports parsing events while it incrementally builds an XML tree. For record-oriented input, handle an "end" event for each completed record: at that point its children and text are available. This example uses an in-memory StringIO fixture containing three <record> elements, collects their identifiers and values, and calls clear() after each record has been read.

The assertions demonstrate that each record produced the expected pair and that the numeric total is 23. element.clear() removes the processed element’s attributes, text, tail, and children, so it can reduce retained detail during this style of processing. It must occur only after extracting what the program needs; clearing earlier would make the child lookup fail. The printed output reports the three records and their aggregate.

iterparse() performs blocking reads from its source. It is therefore not a non-blocking streaming interface, and it does not by itself prove a fixed memory bound for every document shape or surrounding program. Use XMLPullParser when incremental feeding without blocking reads is required. The "end" event used here is available in all supported Python versions; the iterator’s close() method is newer and was added in Python 3.13, but this example does not need it.

AI assistance disclosure: this article was drafted with AI assistance and should be adapted to the document’s record schema.

Source: Python ElementTree documentation.

from io import StringIO
import xml.etree.ElementTree as ET


xml_source = StringIO(
    "<records>"
    '<record id="A"><value>5</value></record>'
    '<record id="B"><value>11</value></record>'
    '<record id="C"><value>7</value></record>'
    "</records>"
)

seen = []
for event, element in ET.iterparse(xml_source, events=("end",)):
    if element.tag != "record":
        continue
    value = int(element.find("value").text)
    seen.append((element.get("id"), value))
    element.clear()

assert seen == [("A", 5), ("B", 11), ("C", 7)]
assert sum(value for _, value in seen) == 23

for record_id, value in seen:
    print(f"record={record_id} value={value}")
print("total=23")

record=A value=5
record=B value=11
record=C value=7
total=23