zlib en Python: comprime datos

El módulo zlib en Python proporciona compresión y descompresión compatibles con la biblioteca zlib y el algoritmo DEFLATE. Trabaja con objetos bytes, permite procesar bloques completos en memoria o flujos por partes y también ofrece checksums Adler-32 y CRC-32.

Usa zlib para protocolos, cachés, almacenamiento interno, bloques binarios y formatos que requieren streams zlib, raw DEFLATE o encapsulado gzip básico. Para leer y escribir archivos .gz con una interfaz de archivo completa, el módulo gzip suele ser más cómodo.

Comprime un bloque en memoria

zlib.compress() recibe bytes y devuelve bytes comprimidos.

import zlib

origen = ("Python y compresión de datos. " * 100).encode("utf-8")
comprimido = zlib.compress(origen)

print(len(origen), len(comprimido))

Los datos repetitivos comprimen bien. El contenido ya comprimido, cifrado o aleatorio puede no reducirse e incluso crecer por el overhead del encabezado.

Descomprime el resultado

restaurado = zlib.decompress(comprimido)
assert restaurado == origen

Un formato inválido, checksum incorrecto, ventana incompatible o stream truncado genera zlib.error. Captura la excepción en la frontera donde entran datos externos.

Elige el nivel de compresión

El argumento level va de 0 a 9, además de -1 para el valor predeterminado.

rapido = zlib.compress(origen, level=zlib.Z_BEST_SPEED)
pequeno = zlib.compress(origen, level=zlib.Z_BEST_COMPRESSION)
predeterminado = zlib.compress(origen, level=zlib.Z_DEFAULT_COMPRESSION)

El nivel 9 usa más CPU para intentar reducir el tamaño. La mejor opción depende de la forma de los datos, almacenamiento, ancho de banda y frecuencia de lectura. Mide con payloads reales.

Comprende wbits

wbits controla la ventana y el envoltorio del stream.

• 9 a 15: encabezado zlib y checksum.
• -9 a -15: raw DEFLATE sin encabezado ni trailer.
• 25 a 31: stream gzip básico.

stream_zlib = zlib.compress(origen, wbits=15)
stream_raw = zlib.compress(origen, wbits=-15)
stream_gzip = zlib.compress(origen, wbits=31)

El descompresor necesita un modo compatible. Pasar raw DEFLATE al modo zlib predeterminado es un error de integración frecuente.

Detecta zlib o gzip automáticamente

Durante la descompresión, valores de wbits entre 40 y 47 aceptan ambos envoltorios.

salida = zlib.decompress(datos_externos, wbits=47)

La autodetección ayuda en integraciones tolerantes. Si el protocolo exige un formato específico, rechaza los demás.

Comprime streams grandes

compressobj() mantiene estado entre bloques y evita cargar todo el archivo en memoria.

compresor = zlib.compressobj(level=6, wbits=15)

with open("entrada.bin", "rb") as origen_archivo, open("entrada.bin.z", "wb") as destino:
    while bloque := origen_archivo.read(64 * 1024):
        destino.write(compresor.compress(bloque))
    destino.write(compresor.flush())

compress() puede retener parte de la entrada internamente. Escribe cada salida y termina siempre con flush().

Finaliza con el modo correcto

El modo predeterminado de flush() es Z_FINISH. Después, el objeto no acepta más datos.

final = compresor.flush(zlib.Z_FINISH)

Z_SYNC_FLUSH mantiene abierto el stream para protocolos interactivos, pero añade marcadores y reduce la eficiencia. Z_FULL_FLUSH reinicia más estado y puede facilitar recuperación, con mayor costo de tamaño.

Descomprime por streaming

descompresor = zlib.decompressobj(wbits=15)

with open("entrada.bin.z", "rb") as origen_archivo, open("restaurado.bin", "wb") as destino:
    while bloque := origen_archivo.read(64 * 1024):
        destino.write(descompresor.decompress(bloque))
    destino.write(descompresor.flush())

if not descompresor.eof:
    raise ValueError("El stream comprimido está incompleto")

eof diferencia un final válido de un archivo truncado. La ausencia de error durante los primeros bloques no es suficiente.

Limita la salida descomprimida

Un payload pequeño puede expandirse a un volumen enorme. Usa max_length y un límite total para reducir el riesgo de decompression bomb.

descompresor = zlib.decompressobj()
limite = 100 * 1024 * 1024
producido = 0
pendiente = payload

while pendiente:
    parte = descompresor.decompress(pendiente, max_length=1024 * 1024)
    producido += len(parte)
    if producido > limite:
        raise ValueError("La salida descomprimida supera el límite")
    guardar(parte)
    pendiente = descompresor.unconsumed_tail
    if not pendiente:
        break

Limita también el tamaño comprimido, tiempo de CPU, cantidad de streams y memoria del proceso. Considera un worker aislado para contenido externo de alto riesgo.

Procesa unconsumed_tail

Cuando max_length impide consumir todo, los bytes pendientes quedan en unconsumed_tail. Debes volver a entregarlos antes de leer más entrada.

Ignorar esta propiedad pierde datos y puede producir una salida parcial sin un error evidente.

Inspecciona unused_data

unused_data contiene bytes situados después del final del stream.

obj = zlib.decompressobj()
salida = obj.decompress(stream_con_sufijo)
salida += obj.flush()
restante = obj.unused_data

Es útil en protocolos que añaden campos después del payload. Valida bytes extra inesperados, porque pueden indicar corrupción o intento de confundir el parser.

Streams concatenados

Un decompressobj termina en el primer stream completo. Si el formato permite varios miembros, crea otro descompresor para unused_data y repite con un límite estricto.

No ejecutes un bucle ilimitado: un atacante podría enviar miles de streams vacíos o diminutos.

Usa diccionarios de compresión

Un zdict mejora mensajes cortos con vocabulario repetido.

diccionario = b'"type":"","id":,"timestamp":,"payload":'
compresor = zlib.compressobj(level=6, zdict=diccionario)
comprimido = compresor.compress(mensaje) + compresor.flush()

descompresor = zlib.decompressobj(zdict=diccionario)
original = descompresor.decompress(comprimido) + descompresor.flush()

Compresor y descompresor deben usar exactamente los mismos bytes. Coloca las secuencias más frecuentes cerca del final del diccionario.

Versiona el diccionario

Un protocolo debe transmitir o negociar un identificador. No pruebes una lista ilimitada de diccionarios, porque aumenta CPU y crea comportamiento ambiguo.

Si usas un bytearray mutable, no lo modifiques entre crear el descompresor y la primera llamada.

Ajusta memLevel y strategy

compressobj() permite controlar memoria y estrategia.

compresor = zlib.compressobj(
    level=6,
    method=zlib.DEFLATED,
    wbits=15,
    memLevel=8,
    strategy=zlib.Z_DEFAULT_STRATEGY,
)

Z_FILTERED puede ayudar datos filtrados, Z_HUFFMAN_ONLY desactiva búsqueda de coincidencias, Z_RLE favorece repeticiones y Z_FIXED usa tablas fijas. Mide antes de cambiar valores predeterminados.

Copia el estado

Compress.copy() permite ramificar salidas que comparten un prefijo comprimido.

base = zlib.compressobj()
prefijo = base.compress(cabecera_comun)

rama_a = base.copy()
salida_a = prefijo + rama_a.compress(payload_a) + rama_a.flush()

rama_b = base.copy()
salida_b = prefijo + rama_b.compress(payload_b) + rama_b.flush()

Los descompresores también pueden copiarse, lo que ayuda en formatos indexados. Documenta el punto exacto del estado.

Calcula CRC-32

checksum = zlib.crc32(datos)
print(f"{checksum:08x}")

Para actualización incremental:

crc = 0
for bloque in bloques:
    crc = zlib.crc32(bloque, crc)

CRC-32 detecta corrupción accidental, pero no autentica. Un atacante puede recalcularlo.

Calcula Adler-32

adler = zlib.adler32(datos)

Adler-32 es rápido para integridad no adversarial. Usa HMAC para integridad autenticada y hashlib para hashes de contenido.

gzip, zipfile y zlib

zlib procesa buffers y streams DEFLATE. gzip es un formato de archivo de un flujo. zipfile almacena varios archivos y metadatos.

Consulta la guía de zipfile en Python, zipapp en Python para aplicaciones y zipimport en Python para imports desde ZIP.

Comprime solo cuando conviene

Los mensajes muy pequeños pueden crecer por el encabezado. Imágenes, vídeo, PDF y ZIP suelen estar comprimidos. Detecta el tipo y compara tamaños antes de guardar otra versión.

Usa bytes y encoding explícito

El módulo no acepta strings.

datos = texto.encode("utf-8")
comprimido = zlib.compress(datos)
restaurado = zlib.decompress(comprimido).decode("utf-8")

El encoding pertenece al protocolo. Define UTF-8 explícitamente y trata errores de decodificación después de validar el stream.

Comprueba versiones

print(zlib.ZLIB_VERSION)
print(zlib.ZLIB_RUNTIME_VERSION)
print(getattr(zlib, "ZLIBNG_VERSION", None))

La biblioteca usada al compilar puede diferir de la cargada en runtime. Python 3.14 expone ZLIBNG_VERSION cuando utiliza zlib-ng. Registra estos valores al investigar incompatibilidades o rendimiento.

Trata zlib.error

try:
    salida = zlib.decompress(payload, wbits=47)
except zlib.error as error:
    registrar_fallo_sin_payload(error)
    raise ValueError("Payload comprimido inválido") from error

No registres los bytes completos. Pueden contener secretos y generar logs enormes.

Seguridad

• Define un máximo de salida descomprimida.
• Aplica límites de CPU y tiempo.
• Comprueba eof para detectar truncamiento.
• Valida unused_data.
• Limita miembros concatenados.
• No uses CRC o Adler para autenticación.
• No confíes en el formato indicado por el usuario.
• Aísla payloads externos de alto riesgo.

Buenas prácticas

• Usa objetos incrementales para datos grandes.
• Mide niveles, estrategias y tamaños de bloque.
• Finaliza compresores con flush().
• Reprocesa unconsumed_tail.
• Verifica eof.
• Versiona los diccionarios.
• Evita recomprimir formatos ya comprimidos.
• Registra versiones zlib en diagnósticos.

Conclusión

zlib en Python ofrece compresión DEFLATE flexible para buffers y streams, con encapsulados zlib, raw y gzip, diccionarios, estrategias, checksums y estado incremental.

Los datos comprimidos externos requieren límites estrictos para evitar expansión excesiva y consumo de recursos. Consulta la documentación oficial de zlib y el manual oficial de la biblioteca zlib.