| """ Python 'zlib_codec' Codec - zlib compression encoding |
| |
| Unlike most of the other codecs which target Unicode, this codec |
| will return Python string objects for both encode and decode. |
| |
| Written by Marc-Andre Lemburg (mal@lemburg.com). |
| |
| """ |
| import codecs |
| import zlib # this codec needs the optional zlib module ! |
| |
| ### Codec APIs |
| |
| def zlib_encode(input,errors='strict'): |
| |
| """ Encodes the object input and returns a tuple (output |
| object, length consumed). |
| |
| errors defines the error handling to apply. It defaults to |
| 'strict' handling which is the only currently supported |
| error handling for this codec. |
| |
| """ |
| assert errors == 'strict' |
| output = zlib.compress(input) |
| return (output, len(input)) |
| |
| def zlib_decode(input,errors='strict'): |
| |
| """ Decodes the object input and returns a tuple (output |
| object, length consumed). |
| |
| input must be an object which provides the bf_getreadbuf |
| buffer slot. Python strings, buffer objects and memory |
| mapped files are examples of objects providing this slot. |
| |
| errors defines the error handling to apply. It defaults to |
| 'strict' handling which is the only currently supported |
| error handling for this codec. |
| |
| """ |
| assert errors == 'strict' |
| output = zlib.decompress(input) |
| return (output, len(input)) |
| |
| class Codec(codecs.Codec): |
| |
| def encode(self, input, errors='strict'): |
| return zlib_encode(input, errors) |
| def decode(self, input, errors='strict'): |
| return zlib_decode(input, errors) |
| |
| class IncrementalEncoder(codecs.IncrementalEncoder): |
| def __init__(self, errors='strict'): |
| assert errors == 'strict' |
| self.errors = errors |
| self.compressobj = zlib.compressobj() |
| |
| def encode(self, input, final=False): |
| if final: |
| c = self.compressobj.compress(input) |
| return c + self.compressobj.flush() |
| else: |
| return self.compressobj.compress(input) |
| |
| def reset(self): |
| self.compressobj = zlib.compressobj() |
| |
| class IncrementalDecoder(codecs.IncrementalDecoder): |
| def __init__(self, errors='strict'): |
| assert errors == 'strict' |
| self.errors = errors |
| self.decompressobj = zlib.decompressobj() |
| |
| def decode(self, input, final=False): |
| if final: |
| c = self.decompressobj.decompress(input) |
| return c + self.decompressobj.flush() |
| else: |
| return self.decompressobj.decompress(input) |
| |
| def reset(self): |
| self.decompressobj = zlib.decompressobj() |
| |
| class StreamWriter(Codec,codecs.StreamWriter): |
| pass |
| |
| class StreamReader(Codec,codecs.StreamReader): |
| pass |
| |
| ### encodings module API |
| |
| def getregentry(): |
| return codecs.CodecInfo( |
| name='zlib', |
| encode=zlib_encode, |
| decode=zlib_decode, |
| incrementalencoder=IncrementalEncoder, |
| incrementaldecoder=IncrementalDecoder, |
| streamreader=StreamReader, |
| streamwriter=StreamWriter, |
| ) |
