write()' callable as specified by PEP 3333
(self, data)
| 19 | error_status = "500 INTERNAL SERVER ERROR" |
| 20 | |
| 21 | def write(self, data): |
| 22 | """'write()' callable as specified by PEP 3333""" |
| 23 | |
| 24 | assert isinstance(data, bytes), "write() argument must be bytestring" |
| 25 | |
| 26 | if not self.status: |
| 27 | raise AssertionError("write() before start_response()") |
| 28 | |
| 29 | elif not self.headers_sent: |
| 30 | # Before the first output, send the stored headers |
| 31 | self.bytes_sent = len(data) # make sure we know content-length |
| 32 | self.send_headers() |
| 33 | else: |
| 34 | self.bytes_sent += len(data) |
| 35 | |
| 36 | # XXX check Content-Length and truncate if too many bytes written? |
| 37 | data = BytesIO(data) |
| 38 | for chunk in iter(lambda: data.read(MAX_SOCKET_CHUNK_SIZE), b""): |
| 39 | self._write(chunk) |
| 40 | self._flush() |
| 41 | |
| 42 | def error_output(self, environ, start_response): |
| 43 | super().error_output(environ, start_response) |
no test coverage detected