This apparently doesn't work correctly. When I tried it, a bunch of log lines were duplicated. π
class LzmaFileHandler(FileHandler):
def _open(self) -> IO[str]:
return lzma.open(
self.baseFilename,
self.mode,
encoding=self.encoding,
errors=self.errors,
)
I had to switch it to this to get it to work:
class CompressionAlgorithm(Enum):
lzma = "lzma"
gzip = "gzip"
bzip2 = "bzip2"
def _compressed_handler(
opener: Callable[[Path, str], IO[str]],
filename: Path,
mode: str = "at",
) -> logging.Handler:
stream = opener(filename, "at")
stream.reconfigure(write_through=True)
atexit.register(stream.close)
return StreamHandler(stream)
COMPRESSION_HANDLERS: Mapping[CompressionAlgorithm, Callable[[Path, str], logging.Handler]] = {
CompressionAlgorithm.lzma: partial(_compressed_handler, lzma.open),
CompressionAlgorithm.gzip: partial(_compressed_handler, gzip.open),
CompressionAlgorithm.bzip2: partial(_compressed_handler, bz2.open),
}
Also,
lzma
buffers a ton in memory before writing, so it didn't really work for my purpose. Even bz2
didn't work, so I had to use gzip
. πdieter_wilhelm likes this.
Tech Cyborg reshared this.