Class that serves as an adapter between TarFile and a stream-like object. The stream-like object only needs to have a read() or write() method that works with bytes, and the method is accessed blockwise. Use of gzip or bzip2 compression is possible. A stream-like
| 325 | os.write(self.fd, s) |
| 326 | |
| 327 | class _Stream: |
| 328 | """Class that serves as an adapter between TarFile and |
| 329 | a stream-like object. The stream-like object only |
| 330 | needs to have a read() or write() method that works with bytes, |
| 331 | and the method is accessed blockwise. |
| 332 | Use of gzip or bzip2 compression is possible. |
| 333 | A stream-like object could be for example: sys.stdin.buffer, |
| 334 | sys.stdout.buffer, a socket, a tape device etc. |
| 335 | |
| 336 | _Stream is intended to be used only internally. |
| 337 | """ |
| 338 | |
| 339 | def __init__(self, name, mode, comptype, fileobj, bufsize, |
| 340 | compresslevel, preset): |
| 341 | """Construct a _Stream object. |
| 342 | """ |
| 343 | self._extfileobj = True |
| 344 | if fileobj is None: |
| 345 | fileobj = _LowLevelFile(name, mode) |
| 346 | self._extfileobj = False |
| 347 | |
| 348 | if comptype == '*': |
| 349 | # Enable transparent compression detection for the |
| 350 | # stream interface |
| 351 | fileobj = _StreamProxy(fileobj) |
| 352 | comptype = fileobj.getcomptype() |
| 353 | |
| 354 | self.name = os.fspath(name) if name is not None else "" |
| 355 | self.mode = mode |
| 356 | self.comptype = comptype |
| 357 | self.fileobj = fileobj |
| 358 | self.bufsize = bufsize |
| 359 | self.buf = b"" |
| 360 | self.pos = 0 |
| 361 | self.closed = False |
| 362 | |
| 363 | try: |
| 364 | if comptype == "gz": |
| 365 | try: |
| 366 | import zlib |
| 367 | except ImportError: |
| 368 | raise CompressionError("zlib module is not available") from None |
| 369 | self.zlib = zlib |
| 370 | self.crc = zlib.crc32(b"") |
| 371 | if mode == "r": |
| 372 | self.exception = zlib.error |
| 373 | self._init_read_gz() |
| 374 | else: |
| 375 | self._init_write_gz(compresslevel) |
| 376 | |
| 377 | elif comptype == "bz2": |
| 378 | try: |
| 379 | import bz2 |
| 380 | except ImportError: |
| 381 | raise CompressionError("bz2 module is not available") from None |
| 382 | if mode == "r": |
| 383 | self.dbuf = b"" |
| 384 | self.cmp = bz2.BZ2Decompressor() |
no outgoing calls
no test coverage detected
searching dependent graphs…