Construct a _Stream object.
(self, name, mode, comptype, fileobj, bufsize,
compresslevel, preset)
| 339 | """ |
| 340 | |
| 341 | def __init__(self, name, mode, comptype, fileobj, bufsize, |
| 342 | compresslevel, preset): |
| 343 | """Construct a _Stream object. |
| 344 | """ |
| 345 | self._extfileobj = True |
| 346 | if fileobj is None: |
| 347 | fileobj = _LowLevelFile(name, mode) |
| 348 | self._extfileobj = False |
| 349 | |
| 350 | if comptype == '*': |
| 351 | # Enable transparent compression detection for the |
| 352 | # stream interface |
| 353 | fileobj = _StreamProxy(fileobj) |
| 354 | comptype = fileobj.getcomptype() |
| 355 | |
| 356 | self.name = os.fspath(name) if name is not None else "" |
| 357 | self.mode = mode |
| 358 | self.comptype = comptype |
| 359 | self.fileobj = fileobj |
| 360 | self.bufsize = bufsize |
| 361 | self.buf = b"" |
| 362 | self.pos = 0 |
| 363 | self.closed = False |
| 364 | |
| 365 | try: |
| 366 | if comptype == "gz": |
| 367 | try: |
| 368 | import zlib |
| 369 | except ImportError: |
| 370 | raise CompressionError("zlib module is not available") from None |
| 371 | self.zlib = zlib |
| 372 | self.crc = zlib.crc32(b"") |
| 373 | if mode == "r": |
| 374 | self.exception = zlib.error |
| 375 | self._init_read_gz() |
| 376 | else: |
| 377 | self._init_write_gz(compresslevel) |
| 378 | |
| 379 | elif comptype == "bz2": |
| 380 | try: |
| 381 | import bz2 |
| 382 | except ImportError: |
| 383 | raise CompressionError("bz2 module is not available") from None |
| 384 | if mode == "r": |
| 385 | self.dbuf = b"" |
| 386 | self.cmp = bz2.BZ2Decompressor() |
| 387 | self.exception = OSError |
| 388 | else: |
| 389 | self.cmp = bz2.BZ2Compressor(compresslevel) |
| 390 | |
| 391 | elif comptype == "xz": |
| 392 | try: |
| 393 | import lzma |
| 394 | except ImportError: |
| 395 | raise CompressionError("lzma module is not available") from None |
| 396 | if mode == "r": |
| 397 | self.dbuf = b"" |
| 398 | self.cmp = lzma.LZMADecompressor() |
no test coverage detected