Construct a _Stream object.
(self, name, mode, comptype, fileobj, bufsize,
compresslevel, preset)
| 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() |
| 385 | self.exception = OSError |
| 386 | else: |
| 387 | self.cmp = bz2.BZ2Compressor(compresslevel) |
| 388 | |
| 389 | elif comptype == "xz": |
| 390 | try: |
| 391 | import lzma |
| 392 | except ImportError: |
| 393 | raise CompressionError("lzma module is not available") from None |
| 394 | if mode == "r": |
| 395 | self.dbuf = b"" |
| 396 | self.cmp = lzma.LZMADecompressor() |
no test coverage detected