1. Home
  2. Articoli
  3. Python

Python e i formati di compressione

La libreria standard di Python copre praticamente ogni esigenza di compressione, e con Python 3.14 anche Zstandard è entrato a farne parte. Le interfacce uniformi rendono facile cambiare formato, ma alcune differenze vanno ricordate: i default dei livelli non sono coerenti tra i moduli, le funzioni in memoria non pongono limiti all'output e l'estrazione degli archivi richiede filtri e controlli espliciti.

Python di Gabriele Romanato

Python è uno dei linguaggi con il supporto alla compressione più ricco nella libreria standard: zlib, gzip, bz2 e lzma per i codec, zipfile e tarfile per gli archivi, shutil per le operazioni di alto livello. Con Python 3.14 si è aggiunto il nuovo package compression, che introduce il supporto nativo a Zstandard e riorganizza i moduli esistenti sotto un unico namespace. In questo articolo vediamo come usare tutti questi strumenti, con particolare attenzione allo streaming e alla sicurezza.

Una panoramica dei moduli

I moduli di compressione di Python condividono un'interfaccia sorprendentemente uniforme, il che rende facile passare da un formato all'altro:

  • funzioni compress() e decompress() per lavorare in memoria;
  • una funzione open() che restituisce un oggetto file, in modalità binaria o testuale, per lavorare con i file su disco;
  • classi compressore e decompressore incrementali (compressobj, BZ2Compressor, LZMACompressor e così via) per lo streaming a basso livello.

I formati coperti sono DEFLATE nelle sue tre varianti (raw, zlib e Gzip), bzip2, LZMA/XZ e, da Python 3.14, Zstandard tramite compression.zstd. Brotli resta invece affidato al pacchetto esterno brotli.

Compressione in memoria

import bz2
import gzip
import json
import lzma
import zlib

payload = json.dumps(
    {"users": [{"id": i, "name": f"user-{i}", "active": i % 2 == 0} for i in range(1000)]}
).encode("utf-8")

results = {
    "zlib": zlib.compress(payload, level=6),
    "gzip": gzip.compress(payload, compresslevel=6),
    "bz2": bz2.compress(payload, compresslevel=9),
    "lzma": lzma.compress(payload, preset=6),
}

print(f"Originale: {len(payload):>8} byte")
for name, data in results.items():
    print(f"{name:<9}: {len(data):>8} byte ({len(data) / len(payload):.1%})")

# Ogni modulo decomprime il proprio formato
assert zlib.decompress(results["zlib"]) == payload
assert gzip.decompress(results["gzip"]) == payload
assert bz2.decompress(results["bz2"]) == payload
assert lzma.decompress(results["lzma"]) == payload

I nomi dei parametri non sono del tutto uniformi: zlib usa level (da 0 a 9, con -1 come default equivalente a 6), gzip e bz2 usano compresslevel (con default 9, attenzione), mentre lzma usa preset (da 0 a 9, eventualmente combinato con lzma.PRESET_EXTREME). Il default di gzip.compress al livello 9 è una trappola frequente: per dati generati al volo il livello 6 offre quasi lo stesso risultato in una frazione del tempo.

Le tre varianti di DEFLATE con zlib

Il modulo zlib gestisce tutte e tre le varianti di DEFLATE tramite il parametro wbits, che codifica sia la dimensione della finestra sia il formato del contenitore:

  • da 9 a 15: formato zlib con intestazione e checksum Adler-32;
  • da -9 a -15: DEFLATE raw senza intestazione, come all'interno dei file ZIP;
  • da 25 a 31 (cioè 16 più la finestra): formato Gzip;
  • da 40 a 47 (cioè 32 più la finestra), solo in decompressione: rilevamento automatico tra zlib e Gzip.
import zlib

data = b"Formati diversi, stesso algoritmo. " * 200


def compress_with(data: bytes, wbits: int, level: int = 6) -> bytes:
    """Comprime data producendo il contenitore indicato da wbits."""
    compressor = zlib.compressobj(level=level, wbits=wbits)
    return compressor.compress(data) + compressor.flush()


raw = compress_with(data, -15)
zlib_format = compress_with(data, 15)
gzip_format = compress_with(data, 31)

# Le dimensioni differiscono solo per intestazione e trailer
print(len(raw), len(zlib_format), len(gzip_format))

# Il rilevamento automatico funziona sia con zlib sia con Gzip
assert zlib.decompress(zlib_format, wbits=47) == data
assert zlib.decompress(gzip_format, wbits=47) == data
# Il formato raw va invece dichiarato esplicitamente
assert zlib.decompress(raw, wbits=-15) == data

# I checksum sono disponibili anche come funzioni indipendenti
print(f"CRC-32: {zlib.crc32(data):08x}, Adler-32: {zlib.adler32(data):08x}")

Lavorare con i file compressi

Le funzioni gzip.open, bz2.open e lzma.open restituiscono oggetti che si comportano come normali file. In modalità testuale ("rt" o "wt") gestiscono anche la codifica dei caratteri, il che permette di leggere un log compresso riga per riga senza mai caricarlo interamente in memoria.

import gzip
import lzma
from collections import Counter
from pathlib import Path


def count_status_codes(log_path: Path) -> Counter[str]:
    """Conta i codici di stato HTTP in un access log compresso in formato combined."""
    counter: Counter[str] = Counter()
    with gzip.open(log_path, "rt", encoding="utf-8", errors="replace") as log:
        for line in log:
            # Nel formato combined il codice di stato segue la richiesta tra virgolette
            parts = line.split('"')
            if len(parts) > 2:
                fields = parts[2].split()
                if fields:
                    counter[fields[0]] += 1
    return counter


def recompress_to_xz(source: Path, destination: Path) -> None:
    """Converte un file Gzip in XZ lavorando a blocchi."""
    with gzip.open(source, "rb") as src, lzma.open(destination, "wb", preset=9) as dst:
        while chunk := src.read(1024 * 1024):
            dst.write(chunk)


if __name__ == "__main__":
    for status, count in count_status_codes(Path("access.log.gz")).most_common():
        print(f"{status}: {count}")
    recompress_to_xz(Path("access.log.gz"), Path("access.log.xz"))

Per copiare un file in un altro senza scrivere il ciclo a mano si può usare shutil.copyfileobj(src, dst, length), che fa esattamente la stessa cosa con un buffer configurabile.

Streaming incrementale

Quando i dati non arrivano da un file ma da un socket, da una coda di messaggi o da una risposta HTTP, servono i compressori incrementali. Si passa un blocco alla volta al metodo compress(), che può restituire zero o più byte di output, e si chiama flush() alla fine per ottenere i dati residui e il trailer.

import zlib
from collections.abc import Iterable, Iterator


def gzip_stream(chunks: Iterable[bytes], level: int = 6) -> Iterator[bytes]:
    """Comprime in formato Gzip un flusso di blocchi, producendo blocchi compressi."""
    compressor = zlib.compressobj(level=level, wbits=31)
    for chunk in chunks:
        if out := compressor.compress(chunk):
            yield out
    yield compressor.flush()


def gunzip_stream(chunks: Iterable[bytes]) -> Iterator[bytes]:
    """Decomprime un flusso Gzip a blocchi."""
    decompressor = zlib.decompressobj(wbits=31)
    for chunk in chunks:
        if out := decompressor.decompress(chunk):
            yield out
    if tail := decompressor.flush():
        yield tail
    if not decompressor.eof:
        raise ValueError("Flusso Gzip troncato")


def generate_rows() -> Iterator[bytes]:
    for i in range(100_000):
        yield f"{i},sensor-{i % 50},{20 + i % 7}.5\n".encode()


compressed = list(gzip_stream(generate_rows()))
restored = b"".join(gunzip_stream(compressed))
print(f"Blocchi compressi: {len(compressed)}, byte decompressi: {len(restored)}")

Il controllo su decompressor.eof è importante: un flusso interrotto a metà non solleva errori durante la decompressione incrementale, semplicemente smette di produrre dati. Solo verificando che sia stata raggiunta la fine del flusso si distingue un trasferimento completo da uno troncato.

Zstandard con Python 3.14

Python 3.14 ha introdotto il package compression (PEP 784), che contiene il nuovo modulo compression.zstd e reesporta i moduli esistenti come compression.gzip, compression.bz2, compression.lzma e compression.zlib. I vecchi nomi restano disponibili e non sono deprecati. Il modulo zstd segue la stessa interfaccia degli altri, con in più il supporto ai dizionari e un controllo fine dei parametri.

from compression import zstd

data = b'{"level":"info","msg":"request handled","status":200}\n' * 5000

# Il livello va da 1 a 22, con valori negativi per modalità ultra veloci; il default è 3
compressed = zstd.compress(data, level=10)
assert zstd.decompress(compressed) == data
print(f"{len(data)} -> {len(compressed)} byte")

# Parametri avanzati: checksum del frame e compressione multithread
options = {
    zstd.CompressionParameter.compression_level: 12,
    zstd.CompressionParameter.checksum_flag: 1,
    zstd.CompressionParameter.nb_workers: 4,
}
with zstd.open("events.jsonl.zst", "wb", options=options) as f:
    f.write(data)

with zstd.open("events.jsonl.zst", "rt", encoding="utf-8") as f:
    print(sum(1 for _ in f), "righe")

Una caratteristica distintiva di Zstandard sono i dizionari: quando si comprimono molti messaggi piccoli e simili tra loro, come record JSON o righe di log inviati singolarmente, un dizionario addestrato su campioni rappresentativi migliora drasticamente il rapporto di compressione, perché fornisce al codec un contesto che il singolo messaggio non ha.

import json
import random

from compression import zstd

# Generiamo campioni realistici di messaggi brevi
samples = [
    json.dumps({
        "event": random.choice(["login", "logout", "purchase", "view"]),
        "user_id": random.randint(1, 10_000),
        "country": random.choice(["IT", "DE", "FR", "ES"]),
        "client": "web",
    }).encode()
    for _ in range(5000)
]

dictionary = zstd.train_dict(samples, 4096)

message = samples[0]
without_dict = zstd.compress(message)
with_dict = zstd.compress(message, zstd_dict=dictionary)

print(f"Messaggio: {len(message)} byte")
print(f"Senza dizionario: {len(without_dict)} byte")
print(f"Con dizionario:   {len(with_dict)} byte")

# Il dizionario deve essere disponibile anche in decompressione
assert zstd.decompress(with_dict, zstd_dict=dictionary) == message

Su versioni di Python precedenti alla 3.14 lo stesso risultato si ottiene con il pacchetto esterno zstandard, che offre un'API diversa ma funzionalmente equivalente.

Brotli

Brotli non fa parte della libreria standard. Il pacchetto ufficiale di Google si installa con pip install brotli e offre un'interfaccia minimale con compress, decompress e una classe Compressor per lo streaming.

import brotli

html = b"<!DOCTYPE html><html><body>" + b"<p>Contenuto ripetuto</p>" * 500 + b"</body></html>"

# quality va da 0 a 11; mode TEXT aiuta il codec con contenuti testuali
compressed = brotli.compress(html, quality=11, mode=brotli.MODE_TEXT)
assert brotli.decompress(compressed) == html
print(f"{len(html)} -> {len(compressed)} byte")

Archivi ZIP con zipfile

Il modulo zipfile supporta la creazione, la lettura e l'aggiunta di voci ad archivi ZIP, con i metodi di compressione ZIP_STORED, ZIP_DEFLATED, ZIP_BZIP2 e ZIP_LZMA (da Python 3.14 anche ZIP_ZSTANDARD, che però non tutti gli strumenti esterni sono in grado di aprire). La classe ZipFile va usata come context manager per garantire che la central directory venga scritta.

import zipfile
from pathlib import Path

# Estensioni di file già compressi, da memorizzare senza ricomprimerli
ALREADY_COMPRESSED = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".mp4", ".zip", ".gz", ".pdf"}


def zip_directory(source: Path, destination: Path) -> None:
    """Crea un archivio ZIP con il contenuto di una directory."""
    with zipfile.ZipFile(destination, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as zf:
        for path in sorted(source.rglob("*")):
            if not path.is_file():
                continue
            arcname = path.relative_to(source).as_posix()
            if path.suffix.lower() in ALREADY_COMPRESSED:
                zf.write(path, arcname, compress_type=zipfile.ZIP_STORED)
            else:
                zf.write(path, arcname)


def list_zip(archive: Path) -> None:
    """Mostra il contenuto di un archivio con il rapporto di compressione di ogni voce."""
    with zipfile.ZipFile(archive) as zf:
        for info in zf.infolist():
            ratio = info.compress_size / info.file_size if info.file_size else 0
            print(f"{info.filename:<40} {info.file_size:>10} {ratio:>7.1%}")
        # testzip verifica il CRC di tutte le voci e restituisce la prima corrotta
        if bad := zf.testzip():
            print(f"Voce corrotta: {bad}")


zip_directory(Path("docs"), Path("docs.zip"))
list_zip(Path("docs.zip"))

Estrazione sicura degli archivi ZIP

Il metodo ZipFile.extractall rimuove già i componenti .. e i percorsi assoluti dai nomi delle voci, quindi è protetto contro lo zip slip classico. Non offre invece alcuna difesa contro le decompression bomb: un archivio di pochi kilobyte può contenere voci che si espandono a decine di gigabyte. La documentazione stessa avverte di non estrarre archivi non fidati senza controlli. Una funzione di estrazione robusta verifica numero di voci, dimensioni dichiarate e rapporto di compressione, e conta i byte effettivamente scritti, poiché le dimensioni nell'intestazione possono essere falsificate.

import shutil
import zipfile
from pathlib import Path


class UnsafeArchiveError(Exception):
    """Archivio che viola i limiti di sicurezza."""


def extract_zip_safely(
    archive: Path,
    destination: Path,
    *,
    max_entries: int = 10_000,
    max_total_size: int = 1024**3,
    max_ratio: float = 200.0,
) -> None:
    destination = destination.resolve()
    destination.mkdir(parents=True, exist_ok=True)

    with zipfile.ZipFile(archive) as zf:
        infos = zf.infolist()
        if len(infos) > max_entries:
            raise UnsafeArchiveError(f"Troppe voci: {len(infos)}")

        written = 0
        for info in infos:
            target = (destination / info.filename).resolve()
            if not target.is_relative_to(destination):
                raise UnsafeArchiveError(f"Percorso non sicuro: {info.filename}")
            if info.compress_size and info.file_size / info.compress_size > max_ratio:
                raise UnsafeArchiveError(f"Rapporto di compressione sospetto: {info.filename}")

            if info.is_dir():
                target.mkdir(parents=True, exist_ok=True)
                continue

            target.parent.mkdir(parents=True, exist_ok=True)
            with zf.open(info) as src, target.open("wb") as dst:
                # Copia a blocchi contando i byte reali, non quelli dichiarati
                while chunk := src.read(64 * 1024):
                    written += len(chunk)
                    if written > max_total_size:
                        raise UnsafeArchiveError("Dimensione totale estratta oltre il limite")
                    dst.write(chunk)


extract_zip_safely(Path("upload.zip"), Path("extracted"))

Archivi TAR e i filtri di estrazione

Il modulo tarfile legge e scrive archivi TAR non compressi o compressi con Gzip, bzip2, XZ e, da Python 3.14, Zstandard. Il formato di compressione si sceglie con la modalità di apertura: "w:gz", "w:bz2", "w:xz", "w:zst", mentre "r:*" rileva automaticamente la compressione in lettura. Per lo streaming da socket o pipe, dove non è possibile spostarsi all'indietro nel file, esistono le modalità con la barra verticale, come "r|gz".

Gli archivi TAR sono storicamente pericolosi perché possono contenere link simbolici, file di dispositivo e permessi arbitrari. Per anni extractall ha estratto tutto senza controlli; Python 3.12 ha introdotto i filtri di estrazione e Python 3.14 ha reso filter="data" il comportamento di default. Questo filtro rifiuta percorsi assoluti, componenti .., link che puntano fuori dalla destinazione e file speciali, e rimuove i bit pericolosi dai permessi. Specificarlo esplicitamente rende il codice sicuro e coerente anche sulle versioni precedenti che lo supportano.

import tarfile
from pathlib import Path


def create_backup(source: Path, archive: Path) -> None:
    """Crea un backup tar.xz escludendo cache e file temporanei."""

    def exclude(info: tarfile.TarInfo) -> tarfile.TarInfo | None:
        name = Path(info.name)
        if "__pycache__" in name.parts or name.suffix in {".tmp", ".pyc"}:
            return None
        # Rimuoviamo le informazioni sull'utente per un archivio riproducibile
        info.uid = info.gid = 0
        info.uname = info.gname = ""
        return info

    with tarfile.open(archive, "w:xz", preset=6) as tar:
        tar.add(source, arcname=source.name, filter=exclude)


def restore_backup(archive: Path, destination: Path) -> None:
    """Estrae un archivio applicando il filtro di sicurezza data."""
    with tarfile.open(archive, "r:*") as tar:
        tar.extractall(destination, filter="data")


create_backup(Path("project"), Path("project.tar.xz"))
restore_backup(Path("project.tar.xz"), Path("restored"))

Il filtro "data" non protegge dalle decompression bomb: per archivi di provenienza esterna conviene comunque iterare sui membri con tar.getmembers(), sommare le dimensioni dichiarate in TarInfo.size e rifiutare l'archivio oltre una soglia prima di estrarlo.

Operazioni di alto livello con shutil

Per i casi più semplici il modulo shutil offre make_archive e unpack_archive, che scelgono il formato in base a un nome ("zip", "tar", "gztar", "bztar", "xztar") o all'estensione del file.

import shutil

# Crea dist/release.tar.gz con il contenuto della directory build
path = shutil.make_archive("dist/release", "gztar", root_dir="build")
print(f"Archivio creato: {path}")

# unpack_archive inoltra il parametro filter all'estrazione dei TAR
shutil.unpack_archive(path, "deploy", filter="data")

Gestire gli errori

Ogni modulo solleva eccezioni specifiche per i dati corrotti: zlib.error, gzip.BadGzipFile, lzma.LZMAError, OSError per bz2, zipfile.BadZipFile e tarfile.TarError. Un file troncato in lettura con gzip.open produce invece un EOFError. Quando si elaborano file provenienti dall'esterno conviene intercettarle tutte in un unico punto e trasformarle in un errore applicativo comprensibile.

import gzip
import zlib


def safe_gunzip(data: bytes, max_size: int = 10 * 1024 * 1024) -> bytes:
    """Decomprime Gzip con un limite sulla dimensione dell'output."""
    decompressor = zlib.decompressobj(wbits=31)
    try:
        # max_length limita l'output; i dati non consumati restano in unconsumed_tail
        result = decompressor.decompress(data, max_size)
    except zlib.error as exc:
        raise ValueError(f"Dati Gzip non validi: {exc}") from exc
    if decompressor.unconsumed_tail:
        raise ValueError(f"Output oltre il limite di {max_size} byte")
    if not decompressor.eof:
        raise ValueError("Flusso Gzip troncato")
    return result


bomb = gzip.compress(b"\0" * (100 * 1024 * 1024))
try:
    safe_gunzip(bomb)
except ValueError as exc:
    print(exc)

Il parametro max_length di decompress è il meccanismo più semplice per difendersi dalle decompression bomb quando si ricevono dati compressi in memoria, ad esempio nel corpo di una richiesta HTTP: la funzione gzip.decompress non offre alcun limite e decomprimerebbe l'intero payload.

Conclusioni

La libreria standard di Python copre praticamente ogni esigenza di compressione, e con Python 3.14 anche Zstandard è entrato a farne parte. Le interfacce uniformi rendono facile cambiare formato, ma alcune differenze vanno ricordate: i default dei livelli non sono coerenti tra i moduli, le funzioni in memoria non pongono limiti all'output e l'estrazione degli archivi richiede filtri e controlli espliciti. Usare filter="data" con i TAR, contare i byte estratti dagli ZIP e sfruttare max_length con i decompressori incrementali sono le tre abitudini che rendono il codice adatto a trattare dati non fidati.