O módulo zlib no Python oferece compressão e descompressão compatíveis com a biblioteca zlib e com o algoritmo DEFLATE. Ele trabalha diretamente com objetos bytes, permite processar dados inteiros em memória ou fluxos por partes e também fornece checksums Adler-32 e CRC-32.
Use zlib para protocolos, caches, armazenamento interno, blocos binários e formatos que exigem streams zlib, raw DEFLATE ou gzip. Para ler e gravar arquivos .gz com cabeçalhos completos, timestamps e interface de arquivo, o módulo gzip normalmente é mais conveniente.
Comprima um bloco em memória
zlib.compress() recebe bytes e devolve bytes comprimidos.
import zlib
texto = ("Python e compressão de dados. " * 100).encode("utf-8")
comprimido = zlib.compress(texto)
print(len(texto), len(comprimido))Dados repetitivos comprimem bem. Conteúdo já comprimido, criptografado ou aleatório pode não diminuir e ainda ganhar alguns bytes de cabeçalho.
Descomprima os dados
restaurado = zlib.decompress(comprimido)
assert restaurado == textoErros de formato, checksum, janela ou stream truncado geram zlib.error. Capture a exceção na fronteira onde dados externos entram na aplicação.
Escolha o nível de compressão
O argumento level varia de 0 a 9, além de -1 para o padrão.
rapido = zlib.compress(texto, level=zlib.Z_BEST_SPEED)
pequeno = zlib.compress(texto, level=zlib.Z_BEST_COMPRESSION)
padrao = zlib.compress(texto, level=zlib.Z_DEFAULT_COMPRESSION)Nível 9 tenta produzir saída menor, mas usa mais CPU. O melhor nível depende do tamanho, repetição dos dados, custo de armazenamento e frequência de leitura. Meça com cargas reais.
Entenda wbits
wbits controla o tamanho da janela e o formato do envelope.
9a15: stream zlib com cabeçalho e checksum.-9a-15: raw DEFLATE sem cabeçalho nem trailer.25a31: stream gzip básico.
stream_zlib = zlib.compress(texto, wbits=15)
stream_raw = zlib.compress(texto, wbits=-15)
stream_gzip = zlib.compress(texto, wbits=31)Quem descomprime precisa usar um modo compatível. Um erro comum é passar raw DEFLATE para o modo zlib padrão.
Aceite zlib ou gzip automaticamente
Na descompressão, valores de 40 a 47 aceitam automaticamente os dois envelopes.
saida = zlib.decompress(dados_externos, wbits=47)Essa flexibilidade ajuda em integrações, mas não substitui a validação do protocolo. Se o contrato exige um formato específico, rejeite os demais.
Comprima streams grandes
compressobj() mantém estado entre blocos e evita carregar toda a entrada na memória.
compressor = zlib.compressobj(level=6, wbits=15)
with open("entrada.bin", "rb") as origem, open("entrada.bin.z", "wb") as destino:
while bloco := origem.read(64 * 1024):
destino.write(compressor.compress(bloco))
destino.write(compressor.flush())compress() pode reter parte da entrada internamente. Sempre escreva a saída de cada chamada e finalize com flush().
Finalize corretamente
O modo padrão de flush() é Z_FINISH. Depois dele, o objeto não aceita novos dados.
final = compressor.flush(zlib.Z_FINISH)Z_SYNC_FLUSH permite continuar comprimindo e é usado em protocolos interativos, mas reduz a eficiência e adiciona marcadores. Z_FULL_FLUSH também reinicia o estado, facilitando recuperação em certos formatos, com custo maior.
Descomprima por streaming
descompressor = zlib.decompressobj(wbits=15)
with open("entrada.bin.z", "rb") as origem, open("restaurado.bin", "wb") as destino:
while bloco := origem.read(64 * 1024):
destino.write(descompressor.decompress(bloco))
destino.write(descompressor.flush())
if not descompressor.eof:
raise ValueError("Stream comprimido incompleto")O atributo eof diferencia um fim válido de um arquivo truncado. Não confie somente na ausência de exceção durante os primeiros blocos.
Limite a saída descomprimida
Dados pequenos podem expandir para volumes enormes. Use max_length e um limite total para reduzir o risco de decompression bomb.
descompressor = zlib.decompressobj()
limite_total = 100 * 1024 * 1024
produzido = 0
pendente = dados
while pendente:
parte = descompressor.decompress(pendente, max_length=1024 * 1024)
produzido += len(parte)
if produzido > limite_total:
raise ValueError("Saída descomprimida excede o limite")
salvar(parte)
pendente = descompressor.unconsumed_tail
if not pendente:
breakTambém limite tamanho comprimido, tempo, quantidade de streams e memória do processo. Para conteúdo não confiável, considere um worker isolado.
Entenda unconsumed_tail
Quando max_length impede o consumo completo, os bytes ainda não processados ficam em unconsumed_tail. Eles precisam ser fornecidos novamente.
Ignorar essa propriedade perde dados e pode produzir um arquivo parcial sem erro óbvio.
Entenda unused_data
unused_data contém bytes após o fim do stream comprimido.
obj = zlib.decompressobj()
saida = obj.decompress(stream_com_sufixo)
saida += obj.flush()
restante = obj.unused_dataIsso é útil quando um protocolo concatena campos após o payload. Valide o tamanho e o conteúdo restante. Bytes extras inesperados podem indicar corrupção ou tentativa de confundir o parser.
Trate streams concatenados
Um decompressobj termina no primeiro stream. Se o formato permite vários membros, crie outro descompressor para unused_data e repita com limite de membros.
Não implemente um loop ilimitado, pois um atacante pode enviar milhares de streams vazios ou pequenos.
Use dicionários de compressão
zdict melhora a compressão de mensagens curtas com vocabulário repetido.
dicionario = b'"type":"","id":,"timestamp":,"payload":'
compressor = zlib.compressobj(level=6, zdict=dicionario)
comprimido = compressor.compress(mensagem) + compressor.flush()
descompressor = zlib.decompressobj(zdict=dicionario)
original = descompressor.decompress(comprimido) + descompressor.flush()Compressor e descompressor precisam usar exatamente os mesmos bytes. Coloque sequências mais frequentes no final do dicionário, conforme recomendação da biblioteca.
Versione o dicionário
Um protocolo deve enviar ou negociar um identificador do dicionário. Nunca tente vários dicionários sem limite, pois isso aumenta custo e pode criar caminhos ambíguos.
Se usar bytearray como dicionário de descompressão, não o modifique entre a criação do objeto e a primeira chamada.
Ajuste memLevel e strategy
compressobj() permite controlar memória e estratégia.
compressor = zlib.compressobj(
level=6,
method=zlib.DEFLATED,
wbits=15,
memLevel=8,
strategy=zlib.Z_DEFAULT_STRATEGY,
)Z_FILTERED pode ajudar dados produzidos por filtros; Z_HUFFMAN_ONLY desativa a busca por correspondências; Z_RLE favorece repetições curtas; Z_FIXED usa códigos Huffman fixos. Faça benchmark antes de alterar.
Copie o estado do compressor
Compress.copy() permite criar saídas alternativas que compartilham um prefixo comum.
base = zlib.compressobj()
prefixo = base.compress(cabecalho_comum)
ramo_a = base.copy()
saida_a = prefixo + ramo_a.compress(payload_a) + ramo_a.flush()
ramo_b = base.copy()
saida_b = prefixo + ramo_b.compress(payload_b) + ramo_b.flush()A mesma ideia existe para descompressão e pode acelerar seeks em um formato indexado. Documente cuidadosamente o estado e os pontos de cópia.
Calcule CRC-32
checksum = zlib.crc32(dados)
print(f"{checksum:08x}")Para atualização incremental:
crc = 0
for bloco in blocos:
crc = zlib.crc32(bloco, crc)CRC-32 detecta erros acidentais, mas não protege contra adulteração maliciosa. Um atacante pode recalcular o checksum.
Calcule Adler-32
adler = zlib.adler32(dados)Adler-32 é rápido e adequado a checksums internos, mas também não é criptográfico. Para integridade autenticada, use HMAC; para hash de conteúdo, use hashlib.
Compare gzip, zipfile e zlib
zlib trabalha com streams e buffers DEFLATE. gzip oferece formato de arquivo para um fluxo. zipfile organiza vários arquivos e metadados em um contêiner.
O guia de zipfile no Python aborda extração segura; zipapp no Python empacota aplicações; e zipimport no Python importa módulos de ZIPs.
Comprima apenas quando vale a pena
Mensagens pequenas podem crescer devido ao envelope. Imagens, vídeos, PDFs e arquivos ZIP geralmente já estão comprimidos. Detecte o tipo e compare o tamanho antes de armazenar a versão comprimida quando o protocolo permitir.
Use bytes e controle encoding
O módulo não recebe strings.
dados = texto.encode("utf-8")
comprimido = zlib.compress(dados)
restaurado = zlib.decompress(comprimido).decode("utf-8")Encoding é responsabilidade da aplicação. Defina UTF-8 explicitamente e trate erros de decodificação depois de validar o stream.
Valide versões da biblioteca
print(zlib.ZLIB_VERSION)
print(zlib.ZLIB_RUNTIME_VERSION)
print(getattr(zlib, "ZLIBNG_VERSION", None))A versão usada na compilação pode diferir da carregada em runtime. Python 3.14 expõe ZLIBNG_VERSION quando construído com zlib-ng. Registre essas informações ao investigar incompatibilidades ou desempenho.
Trate zlib.error
try:
saida = zlib.decompress(payload, wbits=47)
except zlib.error as erro:
registrar_falha_sem_payload(erro)
raise ValueError("Payload comprimido inválido") from erroNão coloque os bytes completos no log. Dados podem conter segredos e produzir logs gigantes.
Segurança
- Defina limite de saída descomprimida.
- Defina timeout e limite de CPU.
- Rejeite streams truncados verificando
eof. - Valide bytes extras em
unused_data. - Limite streams concatenados.
- Não use CRC ou Adler para autenticação.
- Não confie no formato indicado pelo usuário.
- Isole payloads não confiáveis de alto risco.
Boas práticas
- Use streaming para dados grandes.
- Meça nível, estratégia e tamanho de bloco.
- Finalize o compressor com
flush(). - Reprocesse
unconsumed_tail. - Verifique
eof. - Versione dicionários.
- Evite recomprimir formatos prontos.
- Registre versões zlib em diagnósticos.
Conclusão
O zlib no Python fornece acesso flexível ao DEFLATE para buffers e streams, incluindo envelopes zlib, raw e gzip, dicionários, estratégias, checksums e controle incremental.
Compressão de dados externos exige limites rigorosos para evitar expansão excessiva e consumo de recursos. Consulte a documentação oficial do zlib e o manual oficial da biblioteca zlib.
