zlib_codec.py 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677
  1. """Python 'zlib_codec' Codec - zlib compression encoding.
  2. This codec de/encodes from bytes to bytes.
  3. Written by Marc-Andre Lemburg (mal@lemburg.com).
  4. """
  5. import codecs
  6. import zlib # this codec needs the optional zlib module !
  7. ### Codec APIs
  8. def zlib_encode(input, errors='strict'):
  9. assert errors == 'strict'
  10. return (zlib.compress(input), len(input))
  11. def zlib_decode(input, errors='strict'):
  12. assert errors == 'strict'
  13. return (zlib.decompress(input), len(input))
  14. class Codec(codecs.Codec):
  15. def encode(self, input, errors='strict'):
  16. return zlib_encode(input, errors)
  17. def decode(self, input, errors='strict'):
  18. return zlib_decode(input, errors)
  19. class IncrementalEncoder(codecs.IncrementalEncoder):
  20. def __init__(self, errors='strict'):
  21. assert errors == 'strict'
  22. self.errors = errors
  23. self.compressobj = zlib.compressobj()
  24. def encode(self, input, final=False):
  25. if final:
  26. c = self.compressobj.compress(input)
  27. return c + self.compressobj.flush()
  28. else:
  29. return self.compressobj.compress(input)
  30. def reset(self):
  31. self.compressobj = zlib.compressobj()
  32. class IncrementalDecoder(codecs.IncrementalDecoder):
  33. def __init__(self, errors='strict'):
  34. assert errors == 'strict'
  35. self.errors = errors
  36. self.decompressobj = zlib.decompressobj()
  37. def decode(self, input, final=False):
  38. if final:
  39. c = self.decompressobj.decompress(input)
  40. return c + self.decompressobj.flush()
  41. else:
  42. return self.decompressobj.decompress(input)
  43. def reset(self):
  44. self.decompressobj = zlib.decompressobj()
  45. class StreamWriter(Codec, codecs.StreamWriter):
  46. charbuffertype = bytes
  47. class StreamReader(Codec, codecs.StreamReader):
  48. charbuffertype = bytes
  49. ### encodings module API
  50. def getregentry():
  51. return codecs.CodecInfo(
  52. name='zlib',
  53. encode=zlib_encode,
  54. decode=zlib_decode,
  55. incrementalencoder=IncrementalEncoder,
  56. incrementaldecoder=IncrementalDecoder,
  57. streamreader=StreamReader,
  58. streamwriter=StreamWriter,
  59. _is_text_encoding=False,
  60. )