Decode a UTF-8 character split across byte chunks incrementally

By Batu · English technical notes

Also published in our primary archive.

Decode a UTF-8 character split across byte chunks incrementally

How do you decode UTF-8 when a multibyte character is split across chunks? Create one incremental decoder with codecs.getincrementaldecoder("utf-8")() and feed the chunks to that same instance. The fixture splits the three-byte euro sign: b'A\xe2' ends after its first byte, and b'\x82\xacB' supplies the rest.

Decoding each chunk independently under strict UTF-8 fails twice. The first chunk ends with an incomplete leading byte; the second starts with continuation bytes that have no leading byte in that separate operation. The incremental decoder instead emits 'A' after the first call, retains the incomplete suffix internally, and emits '€B' after the second call. Calling decode(b"", final=True) is an explicit flush; here it emits the empty string because no bytes remain buffered.

Incremental decoding is appropriate when chunk boundaries come from a stream rather than from character boundaries. It preserves codec state only in that decoder instance; creating a new decoder per chunk recreates the original failure. A final flush matters because an incomplete sequence at end of input must be handled then, normally by raising under strict errors. The codecs documentation guarantees that the joined incremental output matches decoding the joined input, and describes final=True as flushing the decoder buffer. Python codecs documentation

AI assistance disclosure: This article was drafted with AI assistance and checked against the cited documentation and a synthetic example.

import codecs

first_chunk = b"A\xe2"
second_chunk = b"\x82\xacB"

independent_failures = 0
for chunk in (first_chunk, second_chunk):
    try:
        chunk.decode("utf-8")
    except UnicodeDecodeError:
        independent_failures += 1

assert independent_failures == 2

decoder = codecs.getincrementaldecoder("utf-8")()
first_text = decoder.decode(first_chunk)
second_text = decoder.decode(second_chunk)
final_text = decoder.decode(b"", final=True)

assert (first_text, second_text, final_text) == ("A", "€B", "")
print(f"independent_failures={independent_failures}")
print(f"first={first_text!r}")
print(f"second={second_text!r}")
print(f"flush={final_text!r}")

independent_failures=2
first='A'
second='€B'
flush=''