(self, b)
| 1255 | return self.raw.writable() |
| 1256 | |
| 1257 | def write(self, b): |
| 1258 | if isinstance(b, str): |
| 1259 | raise TypeError("can't write str to binary stream") |
| 1260 | with self._write_lock: |
| 1261 | if self.closed: |
| 1262 | raise ValueError("write to closed file") |
| 1263 | # XXX we can implement some more tricks to try and avoid |
| 1264 | # partial writes |
| 1265 | if len(self._write_buf) > self.buffer_size: |
| 1266 | # We're full, so let's pre-flush the buffer. (This may |
| 1267 | # raise BlockingIOError with characters_written == 0.) |
| 1268 | self._flush_unlocked() |
| 1269 | before = len(self._write_buf) |
| 1270 | self._write_buf.extend(b) |
| 1271 | written = len(self._write_buf) - before |
| 1272 | if len(self._write_buf) > self.buffer_size: |
| 1273 | try: |
| 1274 | self._flush_unlocked() |
| 1275 | except BlockingIOError as e: |
| 1276 | if len(self._write_buf) > self.buffer_size: |
| 1277 | # We've hit the buffer_size. We have to accept a partial |
| 1278 | # write and cut back our buffer. |
| 1279 | overage = len(self._write_buf) - self.buffer_size |
| 1280 | written -= overage |
| 1281 | self._write_buf = self._write_buf[:self.buffer_size] |
| 1282 | raise BlockingIOError(e.errno, e.strerror, written) |
| 1283 | return written |
| 1284 | |
| 1285 | def truncate(self, pos=None): |
| 1286 | with self._write_lock: |
nothing calls this directly
no test coverage detected