Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/multipart.py: 18%

1import base64 

2import binascii 

3import builtins 

4import json 

5import re 

6import sys 

7import uuid 

8import warnings 

9from collections import deque 

10from collections.abc import AsyncIterator, Iterator, Mapping, Sequence 

11from types import TracebackType 

12from typing import TYPE_CHECKING, Any, Union, cast 

13from urllib.parse import parse_qsl, unquote, urlencode 

14 

15from multidict import CIMultiDict 

16 

17from .abc import AbstractStreamWriter 

18from .compression_utils import ZLibCompressor, ZLibDecompressor 

19from .hdrs import ( 

20 CONTENT_DISPOSITION, 

21 CONTENT_ENCODING, 

22 CONTENT_LENGTH, 

23 CONTENT_TRANSFER_ENCODING, 

24 CONTENT_TYPE, 

25) 

26from .helpers import ( 

27 CHAR, 

28 DEFAULT_CHUNK_SIZE, 

29 TOKEN, 

30 HeadersDictProxy, 

31 parse_mimetype, 

32 reify, 

33) 

34from .http import HeadersParser 

35from .http_exceptions import BadHttpMessage 

36from .log import internal_logger 

37from .payload import ( 

38 JsonPayload, 

39 LookupError, 

40 Order, 

41 Payload, 

42 StringPayload, 

43 get_payload, 

44 payload_type, 

45) 

46from .streams import StreamReader 

47 

48if sys.version_info >= (3, 11): 

49 from typing import Self 

50else: 

51 from typing import TypeVar 

52 

53 Self = TypeVar("Self", bound="BodyPartReader") 

54 

55__all__ = ( 

56 "MultipartReader", 

57 "MultipartWriter", 

58 "BodyPartReader", 

59 "BadContentDispositionHeader", 

60 "BadContentDispositionParam", 

61 "parse_content_disposition", 

62 "content_disposition_filename", 

63) 

64 

65 

66# The base64 alphabet plus the padding character. 

67_BASE64_CHARS = frozenset( 

68 b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=" 

69) 

70_NON_BASE64_BYTES = bytes(b for b in range(256) if b not in _BASE64_CHARS) 

71 

72 

73if TYPE_CHECKING: 

74 from .client_reqrep import ClientResponse 

75 

76 

77class BadContentDispositionHeader(RuntimeWarning): 

78 pass 

79 

80 

81class BadContentDispositionParam(RuntimeWarning): 

82 pass 

83 

84 

85def parse_content_disposition( 

86 header: str | None, 

87) -> tuple[str | None, dict[str, str]]: 

88 def is_token(string: str) -> bool: 

89 return bool(string) and TOKEN >= set(string) 

90 

91 def is_quoted(string: str) -> bool: 

92 return len(string) >= 2 and string[0] == string[-1] == '"' 

93 

94 def is_rfc5987(string: str) -> bool: 

95 return is_token(string) and string.count("'") == 2 

96 

97 def is_extended_param(string: str) -> bool: 

98 return string.endswith("*") 

99 

100 def is_continuous_param(string: str) -> bool: 

101 pos = string.find("*") + 1 

102 if not pos: 

103 return False 

104 substring = string[pos:-1] if string.endswith("*") else string[pos:] 

105 return substring.isdigit() 

106 

107 def unescape(text: str, *, chars: str = "".join(map(re.escape, CHAR))) -> str: 

108 return re.sub(f"\\\\([{chars}])", "\\1", text) 

109 

110 if not header: 

111 return None, {} 

112 

113 # https://www.rfc-editor.org/info/rfc9110/#section-5.6.6-2 

114 disptype, *parts = header.split(";") 

115 disptype = disptype.strip() 

116 if not is_token(disptype): 

117 warnings.warn(BadContentDispositionHeader(header)) 

118 return None, {} 

119 

120 params: dict[str, str] = {} 

121 while parts: 

122 item = parts.pop(0) 

123 

124 if not item: # To handle trailing semicolons 

125 warnings.warn(BadContentDispositionHeader(header)) 

126 continue 

127 

128 if "=" not in item: 

129 warnings.warn(BadContentDispositionHeader(header)) 

130 return None, {} 

131 

132 key, value = item.split("=", 1) 

133 key = key.lower().strip() 

134 value = value.lstrip() 

135 

136 if key in params: 

137 warnings.warn(BadContentDispositionHeader(header)) 

138 return None, {} 

139 

140 if not is_token(key): 

141 warnings.warn(BadContentDispositionParam(item)) 

142 continue 

143 

144 elif is_continuous_param(key): 

145 if is_quoted(value): 

146 value = unescape(value[1:-1]) 

147 elif not is_token(value): 

148 warnings.warn(BadContentDispositionParam(item)) 

149 continue 

150 

151 elif is_extended_param(key): 

152 if is_rfc5987(value): 

153 encoding, _, value = value.split("'", 2) 

154 encoding = encoding or "utf-8" 

155 else: 

156 warnings.warn(BadContentDispositionParam(item)) 

157 continue 

158 

159 try: 

160 value = unquote(value, encoding, "strict").lstrip("\\/") 

161 except (builtins.LookupError, UnicodeDecodeError): 

162 # The charset is attacker-controlled here; an unknown name 

163 # raises the builtin LookupError (the bare name is shadowed in 

164 # this module by payload.LookupError). 

165 warnings.warn(BadContentDispositionParam(item)) 

166 continue 

167 

168 else: 

169 failed = True 

170 rstripped = value.rstrip() 

171 if is_quoted(rstripped): 

172 failed = False 

173 value = unescape(rstripped[1:-1].lstrip("\\/")) 

174 elif is_token(value): 

175 failed = False 

176 elif parts: 

177 # maybe just ; in filename, in any case this is just 

178 # one case fix, for proper fix we need to redesign parser 

179 _value = f"{value};{parts[0]}" 

180 if is_quoted(_value): 

181 parts.pop(0) 

182 value = unescape(_value[1:-1].lstrip("\\/")) 

183 failed = False 

184 

185 if failed: 

186 warnings.warn(BadContentDispositionHeader(header)) 

187 return None, {} 

188 

189 params[key] = value 

190 

191 return disptype.lower(), params 

192 

193 

194def content_disposition_filename( 

195 params: Mapping[str, str], name: str = "filename" 

196) -> str | None: 

197 name_suf = "%s*" % name 

198 if not params: 

199 return None 

200 elif name_suf in params: 

201 return params[name_suf] 

202 elif name in params: 

203 return params[name] 

204 else: 

205 # The index is capped at six digits so a header cannot push int() past 

206 # CPython's int-to-str limit; a longer run of digits simply never matches. 

207 section_re = re.compile(re.escape(name) + r"\*([0-9]{1,6})(\*)?") 

208 matches = ( 

209 (m, value) 

210 for key, value in params.items() 

211 if (m := section_re.fullmatch(key)) is not None 

212 ) 

213 # https://www.rfc-editor.org/info/rfc2231/#section-3 

214 # Order numerically. 

215 fnparams = sorted(matches, key=lambda mv: int(mv[0].group(1))) 

216 parts: list[str] = [] 

217 # Consecutive encoded sections are decoded as one unit, because a 

218 # single multibyte character may be split across a section boundary. 

219 pending: list[str] = [] 

220 encoding = "utf-8" 

221 for num, (m, value) in enumerate(fnparams): 

222 if m.group(1) != str(num): # Missing section or leading zero. 

223 return None 

224 if m.group(2) is not None: # encoded parameter 

225 # https://www.rfc-editor.org/info/rfc2231/#section-4.1 

226 if num == 0 and value.count("'") >= 2: 

227 encoding, _, value = value.split("'", 2) 

228 encoding = encoding or "utf-8" 

229 pending.append(value) 

230 continue 

231 if pending: 

232 # Current value is not encoded, so process encoding of previous parts 

233 try: 

234 parts.append(unquote("".join(pending), encoding, "strict")) 

235 except (builtins.LookupError, UnicodeDecodeError): 

236 return None 

237 pending.clear() 

238 parts.append(value) 

239 if pending: 

240 try: 

241 parts.append(unquote("".join(pending), encoding, "strict")) 

242 except (builtins.LookupError, UnicodeDecodeError): 

243 return None 

244 if not parts: 

245 return None 

246 return "".join(parts).lstrip("\\/") 

247 

248 

249class MultipartResponseWrapper: 

250 """Wrapper around the MultipartReader. 

251 

252 It takes care about 

253 underlying connection and close it when it needs in. 

254 """ 

255 

256 def __init__( 

257 self, 

258 resp: "ClientResponse", 

259 stream: "MultipartReader", 

260 ) -> None: 

261 self.resp = resp 

262 self.stream = stream 

263 

264 def __aiter__(self) -> "MultipartResponseWrapper": 

265 return self 

266 

267 async def __anext__( 

268 self, 

269 ) -> Union["MultipartReader", "BodyPartReader"]: 

270 part = await self.next() 

271 if part is None: 

272 raise StopAsyncIteration 

273 return part 

274 

275 def at_eof(self) -> bool: 

276 """Returns True when all response data had been read.""" 

277 return self.resp.content.at_eof() 

278 

279 async def next( 

280 self, 

281 ) -> Union["MultipartReader", "BodyPartReader"] | None: 

282 """Emits next multipart reader object.""" 

283 item = await self.stream.next() 

284 if self.stream.at_eof(): 

285 await self.release() 

286 return item 

287 

288 async def release(self) -> None: 

289 """Release the connection gracefully. 

290 

291 All remaining content is read to the void. 

292 """ 

293 self.resp.release() 

294 

295 

296class BodyPartReader: 

297 """Multipart reader for single body part.""" 

298 

299 chunk_size = 8192 

300 

301 def __init__( 

302 self, 

303 boundary: bytes, 

304 headers: HeadersDictProxy, 

305 content: StreamReader, 

306 *, 

307 subtype: str = "mixed", 

308 default_charset: str | None = None, 

309 max_decompress_size: int = DEFAULT_CHUNK_SIZE, 

310 client_max_size: int = sys.maxsize, 

311 max_size_error_cls: type[Exception] = ValueError, 

312 ) -> None: 

313 self.headers = headers 

314 self._boundary = boundary 

315 self._boundary_len = len(boundary) + 2 # Boundary + \r\n 

316 self._content = content 

317 self._default_charset = default_charset 

318 self._at_eof = False 

319 self._is_form_data = subtype == "form-data" 

320 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8 

321 length = None if self._is_form_data else self.headers.get(CONTENT_LENGTH, None) 

322 if length is not None and not (length.isascii() and length.isdigit()): 

323 # Reject sign prefixes, underscores, whitespace and non-ASCII 

324 # digits that int() would otherwise accept. 

325 # https://www.rfc-editor.org/rfc/rfc9110#section-8.6 

326 raise ValueError(f"invalid Content-Length: {length!r}") 

327 self._length = int(length) if length is not None else None 

328 self._read_bytes = 0 

329 self._b64_carry = b"" 

330 self._unread: deque[bytes] = deque() 

331 self._prev_chunk: bytes | None = None 

332 self._content_eof = 0 

333 self._cache: dict[str, Any] = {} 

334 self._max_decompress_size = max_decompress_size 

335 self._client_max_size = client_max_size 

336 self._max_size_error_cls = max_size_error_cls 

337 

338 def __aiter__(self) -> Self: 

339 return self 

340 

341 async def __anext__(self) -> bytes: 

342 part = await self.next() 

343 if part is None: 

344 raise StopAsyncIteration 

345 return part 

346 

347 async def next(self) -> bytes | None: 

348 item = await self.read() 

349 if not item: 

350 return None 

351 return item 

352 

353 async def read(self, *, decode: bool = False) -> bytes: 

354 """Reads body part data. 

355 

356 decode: Decodes data following by encoding 

357 method from Content-Encoding header. If it missed 

358 data remains untouched 

359 """ 

360 if self._at_eof: 

361 return b"" 

362 data = bytearray() 

363 while not self._at_eof: 

364 data.extend(await self.read_chunk(self.chunk_size)) 

365 if len(data) > self._client_max_size: 

366 raise self._max_size_error_cls(self._client_max_size) 

367 # https://github.com/python/mypy/issues/17537 

368 if decode: # type: ignore[unreachable] 

369 decoded_data = bytearray() 

370 async for d in self.decode_iter(data): 

371 decoded_data.extend(d) 

372 if len(decoded_data) > self._client_max_size: 

373 raise self._max_size_error_cls(self._client_max_size) 

374 return decoded_data 

375 return data 

376 

377 async def read_chunk(self, size: int = chunk_size) -> bytes: 

378 """Reads body part content chunk of the specified size. 

379 

380 size: chunk size 

381 """ 

382 if self._at_eof: 

383 return b"" 

384 carry = self._b64_carry 

385 want = size - len(carry) 

386 if carry: 

387 self._b64_carry = b"" 

388 want = max(want, self._boundary_len) 

389 if self._length: 

390 fresh = await self._read_chunk_from_length(want) 

391 else: 

392 fresh = await self._read_chunk_from_stream(want) 

393 chunk = carry + fresh 

394 self._read_bytes += len(fresh) 

395 

396 # base64 decodes in quartets and every chunk is decoded on its own, so 

397 # a chunk should not end mid-quartet. 

398 encoding = self.headers.get(CONTENT_TRANSFER_ENCODING) 

399 if encoding and encoding.lower() == "base64": 

400 chunk = self._align_base64_chunk(chunk, len(carry) + want) 

401 

402 if self._read_bytes == self._length: 

403 self._at_eof = True 

404 if self._at_eof and await self._content.readline() != b"\r\n": 

405 raise ValueError("Reader did not read all the data or it is malformed") 

406 return chunk 

407 

408 def _align_base64_chunk(self, chunk: bytes, size: int) -> bytes: 

409 at_end = self._at_eof or ( 

410 self._length is not None and self._read_bytes >= self._length 

411 ) 

412 if not at_end and len(chunk) > size: 

413 self._b64_carry = chunk[size:] 

414 chunk = chunk[:size] 

415 

416 remainder = len(chunk.translate(None, _NON_BASE64_BYTES)) % 4 

417 if not remainder or at_end: 

418 return chunk 

419 

420 # Walk back over the trailing partial quartet and carry it into the 

421 # next chunk. 

422 cut = len(chunk) 

423 left = remainder 

424 while left: 

425 cut -= 1 

426 if chunk[cut] in _BASE64_CHARS: 

427 left -= 1 

428 if not cut: 

429 # No whole quartet to hand back, and carrying the lot would make 

430 # no progress: the caller asked for this many bytes, and a part 

431 # that holds no quartet within them holds none to give. 

432 return chunk 

433 

434 self._b64_carry = chunk[cut:] + self._b64_carry 

435 return chunk[:cut] 

436 

437 async def _read_chunk_from_length(self, size: int) -> bytes: 

438 # Reads body part content chunk of the specified size. 

439 # The body part must has Content-Length header with proper value. 

440 assert self._length is not None, "Content-Length required for chunked read" 

441 chunk_size = min(size, self._length - self._read_bytes) 

442 chunk = await self._content.read(chunk_size) 

443 if self._content.at_eof(): 

444 self._at_eof = True 

445 return chunk 

446 

447 async def _read_chunk_from_stream(self, size: int) -> bytes: 

448 # Reads content chunk of body part with unknown length. 

449 # The Content-Length header for body part is not necessary. 

450 assert ( 

451 size >= self._boundary_len 

452 ), "Chunk size must be greater or equal than boundary length + 2" 

453 first_chunk = self._prev_chunk is None 

454 if first_chunk: 

455 # We need to re-add the CRLF that got removed from headers parsing. 

456 self._prev_chunk = b"\r\n" + await self._content.read(size) 

457 

458 chunk = b"" 

459 # content.read() may return less than size, so we need to loop to ensure 

460 # we have enough data to detect the boundary. 

461 while len(chunk) < self._boundary_len: 

462 chunk += await self._content.read(size) 

463 self._content_eof += int(self._content.at_eof()) 

464 if self._content_eof > 2: 

465 raise ValueError("Reading after EOF") 

466 if self._content_eof: 

467 break 

468 if len(chunk) > size: 

469 self._content.unread_data(chunk[size:]) 

470 chunk = chunk[:size] 

471 

472 assert self._prev_chunk is not None 

473 window = self._prev_chunk + chunk 

474 sub = b"\r\n" + self._boundary 

475 if first_chunk: 

476 idx = window.find(sub) 

477 else: 

478 idx = window.find(sub, max(0, len(self._prev_chunk) - len(sub))) 

479 if idx >= 0: 

480 # pushing boundary back to content 

481 with warnings.catch_warnings(): 

482 warnings.filterwarnings("ignore", category=DeprecationWarning) 

483 self._content.unread_data(window[idx:]) 

484 self._prev_chunk = self._prev_chunk[:idx] 

485 chunk = window[len(self._prev_chunk) : idx] 

486 if not chunk: 

487 self._at_eof = True 

488 result = self._prev_chunk[2 if first_chunk else 0 :] # Strip initial CRLF 

489 self._prev_chunk = chunk 

490 return result 

491 

492 async def readline(self) -> bytes: 

493 """Reads body part by line by line.""" 

494 if self._at_eof: 

495 return b"" 

496 

497 if self._unread: 

498 line = self._unread.popleft() 

499 else: 

500 line = await self._content.readline() 

501 

502 if line.startswith(self._boundary): 

503 # the very last boundary may not come with \r\n, 

504 # so set single rules for everyone 

505 sline = line.rstrip(b"\r\n") 

506 boundary = self._boundary 

507 last_boundary = self._boundary + b"--" 

508 # ensure that we read exactly the boundary, not something alike 

509 if sline == boundary or sline == last_boundary: 

510 self._at_eof = True 

511 self._unread.append(line) 

512 return b"" 

513 else: 

514 next_line = await self._content.readline() 

515 if next_line.startswith(self._boundary): 

516 line = line[:-2] # strip CRLF but only once 

517 self._unread.append(next_line) 

518 

519 return line 

520 

521 async def release(self) -> None: 

522 """Like read(), but reads all the data to the void.""" 

523 if self._at_eof: 

524 return 

525 while not self._at_eof: 

526 await self.read_chunk(self.chunk_size) 

527 

528 async def text(self, *, encoding: str | None = None) -> str: 

529 """Like read(), but assumes that body part contains text data.""" 

530 data = await self.read(decode=True) 

531 # see https://www.w3.org/TR/html5/forms.html#multipart/form-data-encoding-algorithm 

532 # and https://dvcs.w3.org/hg/xhr/raw-file/tip/Overview.html#dom-xmlhttprequest-send 

533 encoding = encoding or self.get_charset(default="utf-8") 

534 return data.decode(encoding) 

535 

536 async def json(self, *, encoding: str | None = None) -> dict[str, Any] | None: 

537 """Like read(), but assumes that body parts contains JSON data.""" 

538 data = await self.read(decode=True) 

539 if not data: 

540 return None 

541 encoding = encoding or self.get_charset(default="utf-8") 

542 return cast(dict[str, Any], json.loads(data.decode(encoding))) 

543 

544 async def form(self, *, encoding: str | None = None) -> list[tuple[str, str]]: 

545 """Like read(), but assumes that body parts contain form urlencoded data.""" 

546 data = await self.read(decode=True) 

547 if not data: 

548 return [] 

549 if encoding is not None: 

550 real_encoding = encoding 

551 else: 

552 real_encoding = self.get_charset(default="utf-8") 

553 try: 

554 decoded_data = data.rstrip().decode(real_encoding) 

555 except UnicodeDecodeError: 

556 raise ValueError("data cannot be decoded with %s encoding" % real_encoding) 

557 

558 return parse_qsl( 

559 decoded_data, 

560 keep_blank_values=True, 

561 encoding=real_encoding, 

562 ) 

563 

564 def at_eof(self) -> bool: 

565 """Returns True if the boundary was reached or False otherwise.""" 

566 return self._at_eof 

567 

568 def _apply_content_transfer_decoding(self, data: bytes) -> bytes: 

569 """Apply Content-Transfer-Encoding decoding if header is present.""" 

570 if CONTENT_TRANSFER_ENCODING in self.headers: 

571 return self._decode_content_transfer(data) 

572 return data 

573 

574 def _needs_content_decoding(self) -> bool: 

575 """Check if Content-Encoding decoding should be applied.""" 

576 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8 

577 return not self._is_form_data and CONTENT_ENCODING in self.headers 

578 

579 def decode(self, data: bytes) -> bytes: 

580 """Decodes data synchronously. 

581 

582 Decodes data according the specified Content-Encoding 

583 or Content-Transfer-Encoding headers value. 

584 

585 Note: For large payloads, consider using decode_iter() instead 

586 to avoid blocking the event loop during decompression. 

587 """ 

588 data = self._apply_content_transfer_decoding(data) 

589 if self._needs_content_decoding(): 

590 return self._decode_content(data) 

591 return data 

592 

593 async def decode_iter(self, data: bytes) -> AsyncIterator[bytes]: 

594 """Async generator that yields decoded data chunks. 

595 

596 Decodes data according the specified Content-Encoding 

597 or Content-Transfer-Encoding headers value. 

598 

599 This method offloads decompression to an executor for large payloads 

600 to avoid blocking the event loop. 

601 """ 

602 data = self._apply_content_transfer_decoding(data) 

603 if self._needs_content_decoding(): 

604 async for d in self._decode_content_async(data): 

605 yield d 

606 else: 

607 yield data 

608 

609 def _decode_content(self, data: bytes) -> bytes: 

610 encoding = self.headers.get(CONTENT_ENCODING, "").lower() 

611 if encoding == "identity": 

612 return data 

613 if encoding in {"deflate", "gzip"}: 

614 return ZLibDecompressor( 

615 encoding=encoding, 

616 suppress_deflate_header=True, 

617 ).decompress_sync(data, max_length=self._max_decompress_size) 

618 

619 raise RuntimeError(f"unknown content encoding: {encoding}") 

620 

621 async def _decode_content_async(self, data: bytes) -> AsyncIterator[bytes]: 

622 encoding = self.headers.get(CONTENT_ENCODING, "").lower() 

623 if encoding == "identity": 

624 yield data 

625 elif encoding in {"deflate", "gzip"}: 

626 d = ZLibDecompressor( 

627 encoding=encoding, 

628 suppress_deflate_header=True, 

629 ) 

630 yield await d.decompress(data, max_length=self._max_decompress_size) 

631 while d.data_available: 

632 yield await d.decompress(b"", max_length=self._max_decompress_size) 

633 else: 

634 raise RuntimeError(f"unknown content encoding: {encoding}") 

635 

636 def _decode_content_transfer(self, data: bytes) -> bytes: 

637 encoding = self.headers.get(CONTENT_TRANSFER_ENCODING, "").lower() 

638 

639 if encoding == "base64": 

640 return base64.b64decode(data) 

641 elif encoding == "quoted-printable": 

642 return binascii.a2b_qp(data) 

643 elif encoding in ("binary", "8bit", "7bit"): 

644 return data 

645 else: 

646 raise RuntimeError(f"unknown content transfer encoding: {encoding}") 

647 

648 def get_charset(self, default: str) -> str: 

649 """Returns charset parameter from Content-Type header or default.""" 

650 ctype = self.headers.get(CONTENT_TYPE, "") 

651 mimetype = parse_mimetype(ctype) 

652 return mimetype.parameters.get("charset", self._default_charset or default) 

653 

654 @reify 

655 def name(self) -> str | None: 

656 """Returns name specified in Content-Disposition header. 

657 

658 If the header is missing or malformed, returns None. 

659 """ 

660 _, params = parse_content_disposition(self.headers.get(CONTENT_DISPOSITION)) 

661 return content_disposition_filename(params, "name") 

662 

663 @reify 

664 def filename(self) -> str | None: 

665 """Returns filename specified in Content-Disposition header. 

666 

667 Returns None if the header is missing or malformed. 

668 """ 

669 _, params = parse_content_disposition(self.headers.get(CONTENT_DISPOSITION)) 

670 return content_disposition_filename(params, "filename") 

671 

672 

673@payload_type(BodyPartReader, order=Order.try_first) 

674class BodyPartReaderPayload(Payload): 

675 _value: BodyPartReader 

676 # _autoclose = False (inherited) - Streaming reader that may have resources 

677 

678 def __init__(self, value: BodyPartReader, *args: Any, **kwargs: Any) -> None: 

679 super().__init__(value, *args, **kwargs) 

680 

681 params: dict[str, str] = {} 

682 if value.name is not None: 

683 params["name"] = value.name 

684 if value.filename is not None: 

685 params["filename"] = value.filename 

686 

687 if params: 

688 self.set_content_disposition("attachment", True, **params) 

689 

690 def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str: 

691 raise TypeError("Unable to decode.") 

692 

693 async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes: 

694 """Raises TypeError as body parts should be consumed via write(). 

695 

696 This is intentional: BodyPartReader payloads are designed for streaming 

697 large data (potentially gigabytes) and must be consumed only once via 

698 the write() method to avoid memory exhaustion. They cannot be buffered 

699 in memory for reuse. 

700 """ 

701 raise TypeError("Unable to read body part as bytes. Use write() to consume.") 

702 

703 async def write(self, writer: AbstractStreamWriter) -> None: 

704 field = self._value 

705 # Reading the part drains the underlying stream irreversibly, so mark the 

706 # payload consumed up front: even an interrupted write leaves nothing that 

707 # a retry or redirect could replay. 

708 self._consumed = True 

709 while chunk := await field.read_chunk(size=DEFAULT_CHUNK_SIZE): 

710 async for d in field.decode_iter(chunk): 

711 await writer.write(d) 

712 

713 

714class MultipartReader: 

715 """Multipart body reader.""" 

716 

717 #: Response wrapper, used when multipart readers constructs from response. 

718 response_wrapper_cls = MultipartResponseWrapper 

719 #: Multipart reader class, used to handle multipart/* body parts. 

720 #: None points to type(self) 

721 multipart_reader_cls: type["MultipartReader"] | None = None 

722 #: Body part reader class for non multipart/* content types. 

723 part_reader_cls = BodyPartReader 

724 

725 def __init__( 

726 self, 

727 headers: Mapping[str, str], 

728 content: StreamReader, 

729 *, 

730 client_max_size: int = sys.maxsize, 

731 max_field_size: int = 8190, 

732 max_headers: int = 128, 

733 max_size_error_cls: type[Exception] = ValueError, 

734 ) -> None: 

735 self._mimetype = parse_mimetype(headers[CONTENT_TYPE]) 

736 assert self._mimetype.type == "multipart", "multipart/* content type expected" 

737 if "boundary" not in self._mimetype.parameters: 

738 raise ValueError( 

739 "boundary missed for Content-Type: %s" % headers[CONTENT_TYPE] 

740 ) 

741 

742 self.headers = headers 

743 self._boundary = ("--" + self._get_boundary()).encode() 

744 self._client_max_size = client_max_size 

745 self._content = content 

746 self._default_charset: str | None = None 

747 self._last_part: MultipartReader | BodyPartReader | None = None 

748 self._max_field_size = max_field_size 

749 self._max_headers = max_headers 

750 self._max_size_error_cls = max_size_error_cls 

751 self._at_eof = False 

752 self._at_bof = True 

753 self._unread: list[bytes] = [] 

754 

755 def __aiter__(self) -> Self: 

756 return self 

757 

758 async def __anext__( 

759 self, 

760 ) -> Union["MultipartReader", BodyPartReader] | None: 

761 part = await self.next() 

762 if part is None: 

763 raise StopAsyncIteration 

764 return part 

765 

766 @classmethod 

767 def from_response( 

768 cls, 

769 response: "ClientResponse", 

770 ) -> MultipartResponseWrapper: 

771 """Constructs reader instance from HTTP response. 

772 

773 :param response: :class:`~aiohttp.client.ClientResponse` instance 

774 """ 

775 obj = cls.response_wrapper_cls( 

776 response, cls(response.headers, response.content) 

777 ) 

778 return obj 

779 

780 def at_eof(self) -> bool: 

781 """Returns True if the final boundary was reached, false otherwise.""" 

782 return self._at_eof 

783 

784 async def next( 

785 self, 

786 ) -> Union["MultipartReader", BodyPartReader] | None: 

787 """Emits the next multipart body part.""" 

788 # So, if we're at BOF, we need to skip till the boundary. 

789 if self._at_eof: 

790 return None 

791 await self._maybe_release_last_part() 

792 if self._at_bof: 

793 await self._read_until_first_boundary() 

794 self._at_bof = False 

795 else: 

796 await self._read_boundary() 

797 if self._at_eof: # we just read the last boundary, nothing to do there 

798 # https://github.com/python/mypy/issues/17537 

799 return None # type: ignore[unreachable] 

800 

801 part = await self.fetch_next_part() 

802 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.6 

803 if (