Minimal object to pass WriteBuffer flushes into GzipFile
| 127 | |
| 128 | |
| 129 | class _WriteBufferStream(io.RawIOBase): |
| 130 | """Minimal object to pass WriteBuffer flushes into GzipFile""" |
| 131 | def __init__(self, gzip_file): |
| 132 | self.gzip_file = weakref.ref(gzip_file) |
| 133 | |
| 134 | def write(self, data): |
| 135 | gzip_file = self.gzip_file() |
| 136 | if gzip_file is None: |
| 137 | raise RuntimeError("lost gzip_file") |
| 138 | return gzip_file._write_raw(data) |
| 139 | |
| 140 | def seekable(self): |
| 141 | return False |
| 142 | |
| 143 | def writable(self): |
| 144 | return True |
| 145 | |
| 146 | |
| 147 | class GzipFile(_streams.BaseStream): |
no outgoing calls
no test coverage detected
searching dependent graphs…