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