(self, b)
| 1236 | return self.raw.writable() |
| 1237 | |
| 1238 | def write(self, b): |
| 1239 | if isinstance(b, str): |
| 1240 | raise TypeError("can't write str to binary stream") |
| 1241 | with self._write_lock: |
| 1242 | if self.closed: |
| 1243 | raise ValueError("write to closed file") |
| 1244 | # XXX we can implement some more tricks to try and avoid |
| 1245 | # partial writes |
| 1246 | if len(self._write_buf) > self.buffer_size: |
| 1247 | # We're full, so let's pre-flush the buffer. (This may |
| 1248 | # raise BlockingIOError with characters_written == 0.) |
| 1249 | self._flush_unlocked() |
| 1250 | before = len(self._write_buf) |
| 1251 | self._write_buf.extend(b) |
| 1252 | written = len(self._write_buf) - before |
| 1253 | if len(self._write_buf) > self.buffer_size: |
| 1254 | try: |
| 1255 | self._flush_unlocked() |
| 1256 | except BlockingIOError as e: |
| 1257 | if len(self._write_buf) > self.buffer_size: |
| 1258 | # We've hit the buffer_size. We have to accept a partial |
| 1259 | # write and cut back our buffer. |
| 1260 | overage = len(self._write_buf) - self.buffer_size |
| 1261 | written -= overage |
| 1262 | self._write_buf = self._write_buf[:self.buffer_size] |
| 1263 | raise BlockingIOError(e.errno, e.strerror, written) |
| 1264 | return written |
| 1265 | |
| 1266 | def truncate(self, pos=None): |
| 1267 | with self._write_lock: |
nothing calls this directly
no test coverage detected