0
# Content Decoders
1
2
Streaming decoders for Base64 and quoted-printable encoded content with automatic caching for incomplete chunks. These decoders enable processing of encoded form data without loading entire payloads into memory.
3
4
## Protocol Types
5
6
```python { .api }
7
class SupportsWrite(Protocol):
8
def write(self, __b: bytes) -> object: ...
9
```
10
11
## Capabilities
12
13
### Base64Decoder
14
15
Provides interface to decode a stream of Base64 data with automatic caching for arbitrary-sized writes and proper handling of incomplete chunks.
16
17
```python { .api }
18
class Base64Decoder:
19
"""
20
Streaming Base64 decoder with chunk caching.
21
"""
22
23
def __init__(self, underlying: SupportsWrite[bytes]) -> None:
24
"""
25
Initialize Base64Decoder.
26
27
Parameters:
28
- underlying: Object with write() method to receive decoded data
29
"""
30
31
def write(self, data: bytes) -> int:
32
"""
33
Decode Base64 data and write to underlying object.
34
35
Parameters:
36
- data: Base64 encoded bytes to decode
37
38
Returns:
39
Number of input bytes processed
40
41
Raises:
42
- DecodeError: If invalid Base64 data is encountered
43
"""
44
45
def close(self) -> None:
46
"""
47
Close decoder and underlying object if it has close() method.
48
"""
49
50
def finalize(self) -> None:
51
"""
52
Finalize decoder, writing any remaining cached data.
53
54
Raises:
55
- DecodeError: If data remains in cache (incomplete Base64)
56
"""
57
58
# Properties
59
cache: bytearray # Cache for incomplete Base64 chunks
60
underlying: SupportsWrite[bytes] # Underlying object to write decoded data
61
```
62
63
**Usage Example:**
64
65
```python
66
from python_multipart.decoders import Base64Decoder
67
from python_multipart.exceptions import DecodeError
68
import base64
69
import io
70
71
def decode_base64_stream(encoded_stream, output_file):
72
"""Decode Base64 stream to file."""
73
74
with open(output_file, 'wb') as f:
75
decoder = Base64Decoder(f)
76
77
try:
78
# Process stream in chunks
79
while True:
80
chunk = encoded_stream.read(1024)
81
if not chunk:
82
break
83
decoder.write(chunk)
84
85
# Finalize to flush any remaining data
86
decoder.finalize()
87
88
except DecodeError as e:
89
print(f"Base64 decode error: {e}")
90
raise
91
finally:
92
decoder.close()
93
94
# Example with in-memory decoding
95
def decode_base64_to_memory(base64_data):
96
"""Decode Base64 data to memory buffer."""
97
98
output_buffer = io.BytesIO()
99
decoder = Base64Decoder(output_buffer)
100
101
try:
102
# Can handle partial chunks
103
chunk_size = 100
104
for i in range(0, len(base64_data), chunk_size):
105
chunk = base64_data[i:i + chunk_size]
106
decoder.write(chunk)
107
108
decoder.finalize()
109
110
# Get decoded data
111
output_buffer.seek(0)
112
return output_buffer.read()
113
114
except DecodeError as e:
115
print(f"Decode error: {e}")
116
return None
117
finally:
118
decoder.close()
119
120
# Test with sample data
121
original_data = b"Hello, World! This is a test message."
122
encoded_data = base64.b64encode(original_data)
123
124
print(f"Original: {original_data}")
125
print(f"Encoded: {encoded_data}")
126
127
decoded_data = decode_base64_to_memory(encoded_data)
128
print(f"Decoded: {decoded_data}")
129
print(f"Match: {original_data == decoded_data}")
130
```
131
132
**Handling Incomplete Data:**
133
134
```python
135
from python_multipart.decoders import Base64Decoder
136
from python_multipart.exceptions import DecodeError
137
import io
138
139
def demonstrate_chunk_handling():
140
"""Show how decoder handles incomplete Base64 chunks."""
141
142
# Base64 data that doesn't align to 4-byte boundaries
143
base64_data = b"SGVsbG8gV29ybGQh" # "Hello World!" encoded
144
145
output = io.BytesIO()
146
decoder = Base64Decoder(output)
147
148
# Feed data in unaligned chunks
149
chunks = [
150
base64_data[:3], # "SGV" - incomplete
151
base64_data[3:7], # "sbG8" - complete group
152
base64_data[7:10], # "gV2" - incomplete
153
base64_data[10:] # "9ybGQh" - remainder
154
]
155
156
try:
157
for i, chunk in enumerate(chunks):
158
print(f"Writing chunk {i}: {chunk}")
159
decoder.write(chunk)
160
print(f"Cache after chunk {i}: {decoder.cache}")
161
162
decoder.finalize()
163
164
# Get result
165
output.seek(0)
166
result = output.read()
167
print(f"Decoded result: {result}")
168
169
except DecodeError as e:
170
print(f"Error: {e}")
171
finally:
172
decoder.close()
173
174
demonstrate_chunk_handling()
175
```
176
177
### QuotedPrintableDecoder
178
179
Provides interface to decode a stream of quoted-printable data with caching for incomplete escape sequences.
180
181
```python { .api }
182
class QuotedPrintableDecoder:
183
"""
184
Streaming quoted-printable decoder with chunk caching.
185
"""
186
187
def __init__(self, underlying: SupportsWrite[bytes]) -> None:
188
"""
189
Initialize QuotedPrintableDecoder.
190
191
Parameters:
192
- underlying: Object with write() method to receive decoded data
193
"""
194
195
def write(self, data: bytes) -> int:
196
"""
197
Decode quoted-printable data and write to underlying object.
198
199
Parameters:
200
- data: Quoted-printable encoded bytes to decode
201
202
Returns:
203
Number of input bytes processed
204
"""
205
206
def close(self) -> None:
207
"""
208
Close decoder and underlying object if it has close() method.
209
"""
210
211
def finalize(self) -> None:
212
"""
213
Finalize decoder, writing any remaining cached data.
214
Does not raise exceptions for incomplete data.
215
"""
216
217
# Properties
218
cache: bytes # Cache for incomplete quoted-printable chunks
219
underlying: SupportsWrite[bytes] # Underlying object to write decoded data
220
```
221
222
**Usage Example:**
223
224
```python
225
from python_multipart.decoders import QuotedPrintableDecoder
226
import binascii
227
import io
228
229
def decode_quoted_printable_stream(encoded_stream, output_file):
230
"""Decode quoted-printable stream to file."""
231
232
with open(output_file, 'wb') as f:
233
decoder = QuotedPrintableDecoder(f)
234
235
try:
236
while True:
237
chunk = encoded_stream.read(1024)
238
if not chunk:
239
break
240
decoder.write(chunk)
241
242
decoder.finalize()
243
244
finally:
245
decoder.close()
246
247
# Example with email-style quoted-printable content
248
def decode_email_content():
249
"""Decode typical email quoted-printable content."""
250
251
# Sample quoted-printable data
252
qp_data = b"Hello=20World!=0D=0AThis=20is=20a=20test."
253
# Decodes to: "Hello World!\r\nThis is a test."
254
255
output = io.BytesIO()
256
decoder = QuotedPrintableDecoder(output)
257
258
# Process in small chunks to test caching
259
chunk_size = 5
260
for i in range(0, len(qp_data), chunk_size):
261
chunk = qp_data[i:i + chunk_size]
262
print(f"Processing chunk: {chunk}")
263
decoder.write(chunk)
264
265
decoder.finalize()
266
267
# Get decoded result
268
output.seek(0)
269
result = output.read()
270
print(f"Decoded: {result}")
271
return result
272
273
decode_email_content()
274
```
275
276
**Handling Escape Sequences:**
277
278
```python
279
from python_multipart.decoders import QuotedPrintableDecoder
280
import io
281
282
def demonstrate_escape_handling():
283
"""Show how decoder handles incomplete escape sequences."""
284
285
# Quoted-printable with escape sequences split across chunks
286
qp_data = b"Hello=3DWorld=21=0AEnd"
287
# Should decode to: "Hello=World!\nEnd"
288
289
output = io.BytesIO()
290
decoder = QuotedPrintableDecoder(output)
291
292
# Split data to break escape sequences
293
chunks = [
294
b"Hello=3", # Incomplete escape
295
b"DWorld=2", # Complete + incomplete
296
b"1=0AEnd" # Complete sequences
297
]
298
299
for i, chunk in enumerate(chunks):
300
print(f"Chunk {i}: {chunk}")
301
decoder.write(chunk)
302
print(f"Cache after chunk {i}: {decoder.cache}")
303
304
decoder.finalize()
305
306
output.seek(0)
307
result = output.read()
308
print(f"Final result: {result}")
309
print(f"As string: {result.decode('utf-8')}")
310
311
demonstrate_escape_handling()
312
```
313
314
### Integration with Form Parsing
315
316
Decoders are typically used internally by the parsing system when Content-Transfer-Encoding headers specify encoded content:
317
318
```python
319
from python_multipart import FormParser
320
from python_multipart.decoders import Base64Decoder, QuotedPrintableDecoder
321
import io
322
323
def handle_encoded_form_data():
324
"""Example of how decoders integrate with form parsing."""
325
326
# This is typically handled automatically by FormParser
327
# but shown here for illustration
328
329
def create_decoder_for_encoding(encoding, output):
330
"""Create appropriate decoder based on encoding type."""
331
if encoding.lower() == 'base64':
332
return Base64Decoder(output)
333
elif encoding.lower() == 'quoted-printable':
334
return QuotedPrintableDecoder(output)
335
else:
336
return output # No decoding needed
337
338
def process_encoded_part(content_transfer_encoding, data):
339
"""Process a form part with content encoding."""
340
341
output = io.BytesIO()
342
343
if content_transfer_encoding:
344
decoder = create_decoder_for_encoding(content_transfer_encoding, output)
345
346
# Write encoded data through decoder
347
decoder.write(data)
348
decoder.finalize()
349
350
if hasattr(decoder, 'close'):
351
decoder.close()
352
else:
353
# No encoding - write directly
354
output.write(data)
355
356
# Get decoded result
357
output.seek(0)
358
return output.read()
359
360
# Example usage
361
base64_data = b"SGVsbG8gV29ybGQh" # "Hello World!" in Base64
362
qp_data = b"Hello=20World=21" # "Hello World!" in quoted-printable
363
364
decoded_b64 = process_encoded_part('base64', base64_data)
365
decoded_qp = process_encoded_part('quoted-printable', qp_data)
366
367
print(f"Base64 decoded: {decoded_b64}")
368
print(f"QP decoded: {decoded_qp}")
369
370
handle_encoded_form_data()
371
```
372
373
### Custom Decoder Usage
374
375
Decoders can be used independently for any streaming decode operation:
376
377
```python
378
from python_multipart.decoders import Base64Decoder
379
import io
380
381
class DataProcessor:
382
"""Custom data processor that uses Base64Decoder."""
383
384
def __init__(self):
385
self.processed_data = io.BytesIO()
386
self.decoder = Base64Decoder(self.processed_data)
387
388
def process_chunk(self, encoded_chunk):
389
"""Process a chunk of Base64 encoded data."""
390
return self.decoder.write(encoded_chunk)
391
392
def get_result(self):
393
"""Get the final decoded result."""
394
self.decoder.finalize()
395
self.processed_data.seek(0)
396
result = self.processed_data.read()
397
self.decoder.close()
398
return result
399
400
# Usage
401
processor = DataProcessor()
402
processor.process_chunk(b"SGVsbG8g")
403
processor.process_chunk(b"V29ybGQh")
404
result = processor.get_result()
405
print(f"Processed result: {result}") # b"Hello World!"
406
```