(self, src, headers_section, data_section, max_read)
| 1455 | self.headers = {} |
| 1456 | |
| 1457 | def read(self, src, headers_section, data_section, max_read): |
| 1458 | start, end = headers_section |
| 1459 | sz = end - start |
| 1460 | has_read = sz |
| 1461 | if has_read > max_read: |
| 1462 | raise HTTPError(413, 'Request entity too large') |
| 1463 | src.seek(start) |
| 1464 | headers_raw = tonat(src.read(sz)) |
| 1465 | for header_raw in headers_raw.splitlines(): |
| 1466 | header = self.parse_header(header_raw) |
| 1467 | self.headers[header.name] = header |
| 1468 | if header.name == 'Content-Disposition': |
| 1469 | self.name = header.options['name'] |
| 1470 | self.filename = header.options.get('filename') |
| 1471 | elif header.name == 'Content-Type': |
| 1472 | self.ctype = header.value |
| 1473 | if self.name is None: |
| 1474 | raise HTTPError(422, 'Noname field found while parsing multipart/formdata body: %s' % header_raw) |
| 1475 | if self.filename is not None: |
| 1476 | self.file = MPBytesIOProxy(src, *data_section) |
| 1477 | else: |
| 1478 | start, end = data_section |
| 1479 | sz = end - start |
| 1480 | if sz: |
| 1481 | has_read += sz |
| 1482 | if has_read > max_read: |
| 1483 | raise HTTPError(413, 'Request entity too large') |
| 1484 | src.seek(start) |
| 1485 | self.value = tonat(src.read(sz)) |
| 1486 | else: |
| 1487 | self.value = '' |
| 1488 | return has_read |
| 1489 | |
| 1490 | @classmethod |
| 1491 | def parse_header(cls, s): |
nothing calls this directly
no test coverage detected