| 1290 | |
| 1291 | |
| 1292 | class _ZipWriteFile(io.BufferedIOBase): |
| 1293 | def __init__(self, zf, zinfo, zip64): |
| 1294 | self._zinfo = zinfo |
| 1295 | self._zip64 = zip64 |
| 1296 | self._zipfile = zf |
| 1297 | self._compressor = _get_compressor(zinfo.compress_type, |
| 1298 | zinfo.compress_level) |
| 1299 | self._file_size = 0 |
| 1300 | self._compress_size = 0 |
| 1301 | self._crc = 0 |
| 1302 | |
| 1303 | @property |
| 1304 | def _fileobj(self): |
| 1305 | return self._zipfile.fp |
| 1306 | |
| 1307 | @property |
| 1308 | def name(self): |
| 1309 | return self._zinfo.filename |
| 1310 | |
| 1311 | @property |
| 1312 | def mode(self): |
| 1313 | return 'wb' |
| 1314 | |
| 1315 | def writable(self): |
| 1316 | return True |
| 1317 | |
| 1318 | def write(self, data): |
| 1319 | if self.closed: |
| 1320 | raise ValueError('I/O operation on closed file.') |
| 1321 | |
| 1322 | # Accept any data that supports the buffer protocol |
| 1323 | if isinstance(data, (bytes, bytearray)): |
| 1324 | nbytes = len(data) |
| 1325 | else: |
| 1326 | data = memoryview(data) |
| 1327 | nbytes = data.nbytes |
| 1328 | self._file_size += nbytes |
| 1329 | |
| 1330 | self._crc = crc32(data, self._crc) |
| 1331 | if self._compressor: |
| 1332 | data = self._compressor.compress(data) |
| 1333 | self._compress_size += len(data) |
| 1334 | self._fileobj.write(data) |
| 1335 | return nbytes |
| 1336 | |
| 1337 | def close(self): |
| 1338 | if self.closed: |
| 1339 | return |
| 1340 | try: |
| 1341 | super().close() |
| 1342 | # Flush any data from the compressor, and update header info |
| 1343 | if self._compressor: |
| 1344 | buf = self._compressor.flush() |
| 1345 | self._compress_size += len(buf) |
| 1346 | self._fileobj.write(buf) |
| 1347 | self._zinfo.compress_size = self._compress_size |
| 1348 | else: |
| 1349 | self._zinfo.compress_size = self._file_size |
no outgoing calls
no test coverage detected
searching dependent graphs…