Extract a .zst file in Python

README.md

Extract Zstd archives with Python

Uses compression.zstd (Python 3.14+) or zstandard (Python < 3.14) to extract Zstandard-compressed files and archives.

• .zst files (raw Zstandard-compressed single files)
• .tar.zst archives
• mislabeled plain tar files (for example, a .tar file accidentally named .zst)

The extractor is implemented in extract_zst.py.

Streams decompression to not overflow memory for large files. Uses safer tar extraction filtering on supported Python versions.

Requirements

• Python 3.9+
• For Python < 3.14: zstandard

The script includes inline metadata for uv run --script users.

uv run --script extract_zst.py

Usage

Run with Python:

python .\extract_zst.py <archive.zst> <output_directory>

Examples:

python .\extract_zst.py .\hello.txt.zst .
python .\extract_zst.py .\test.tar.zst .\out

Create A Proper .tar.zst

A real .tar.zst is a tar stream compressed with Zstandard.

If your tar supports zstd directly:

tar --zstd -cf test.tar.zst hello.txt goodbye.txt

Portable two-step approach:

tar -cf test.tar hello.txt goodbye.txt
zstd -19 -T0 --rm test.tar -o test.tar.zst

Quick validation:

zstd -t test.tar.zst
tar --zstd -tf test.tar.zst

Running Tests (PyTest)

This repository uses PyTest.

pytest

Project Files

• extract_zst.py: extractor implementation and CLI entrypoint
• test_extract_zst.py: pytest test suite
• hello.txt.zst, test.tar.zst: sample archives for local testing

extract_zst.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = ["zstandard; python_version < '3.14'"]
# ///

"""
Zstandard (.zst / .tar.zst) extractor.

Works on Python >= 3.9 (with zstandard on < 3.14).
Handles both plain .zst files and .tar.zst archives.
Uses tarfile's built-in Zstd support on Python 3.14+ when possible.

Reference: https://docs.python.org/3.14/library/compression.zstd.html#compression.zstd.ZstdFile
"""

from pathlib import Path
import sys
import tarfile
import shutil
import tempfile
import argparse

if sys.version_info >= (3, 14):
    import compression.zstd as zstd  # type: ignore
else:
    import zstandard as zstd  # type: ignore


def _extract_tar(tar: tarfile.TarFile, out_path: Path) -> None:
    """Extract tar contents with a safe filter when supported by Python."""
    if sys.version_info >= (3, 12):
        tar.extractall(out_path, filter="data")
    else:
        tar.extractall(out_path)


def extract_zstd(archive: Path | str, out_path: Path | str) -> None:
    """Extract a .zst (Zstandard) file, handling both plain compressed files
    and .tar.zst archives.

    - If the decompressed content is a valid tar archive, it extracts the
      contents to `out_path` (directory).
    - Otherwise, it decompresses the content directly to a file named after
      the archive (minus .zst extension) inside `out_path`.

    Works cross-platform and streams for large files.
    """
    archive = Path(archive).expanduser()
    out_path = Path(out_path).expanduser().resolve()
    out_path.mkdir(parents=True, exist_ok=True)

    if not archive.is_file():
        raise FileNotFoundError(f"Archive not found: {archive}")

    # Some inputs are mislabeled as .zst but are already plain tar archives.
    if tarfile.is_tarfile(archive):
        with tarfile.open(archive, mode="r:*") as tar:
            _extract_tar(tar, out_path)
        print(f"Extracted tar archive to {out_path}")
        return

    # Prefer tarfile's native Zstd support on Python 3.14+
    if sys.version_info >= (3, 14):
        try:
            # Try opening as tar.zst directly (most common case)
            with tarfile.open(archive, mode="r:zst") as tar:
                _extract_tar(tar, out_path)
            print(f"Extracted tar.zst archive to {out_path}")
            return
        except tarfile.ReadError:
            # Not a tar file — fall back to raw decompression
            pass

    lower_name = archive.name.lower()
    tar_zst_suffixes = (".tar.zst", ".tar.zstd", ".tzst", ".tzstd")

    # Fallback for tar.zst on paths where native tarfile zstd support is unavailable.
    # Use a temporary tar file so no intermediate artifact is left in out_path.
    if lower_name.endswith(tar_zst_suffixes):
        with tempfile.TemporaryDirectory(prefix="zstd_extract_") as tmp_dir:
            temp_tar = Path(tmp_dir) / archive.with_suffix("").name
            _decompress_zstd(archive, temp_tar)
            try:
                with tarfile.open(temp_tar, mode="r:*") as tar:
                    _extract_tar(tar, out_path)
                print(f"Extracted tar.zst archive to {out_path}")
                return
            except tarfile.ReadError:
                # If decompressed data is not tar, fall through to raw output path.
                pass

    # Legacy path or raw .zst (non-tar)
    out_file = _build_out_file(archive, out_path)
    _decompress_zstd(archive, out_file)
    print(f"Decompressed {archive.name} -> {out_file}")


def _build_out_file(archive: Path, out_path: Path) -> Path:
    """Compute output path for raw Zstd decompression."""
    # Determine output filename (strip .zst or .tzst etc.)
    if archive.suffix.lower() in {".zst", ".zstd"}:
        return out_path / archive.with_suffix("").name
    else:
        return out_path / archive.name


def _decompress_zstd(archive: Path, out_file: Path) -> None:
    """Internal helper for Zstd decompression to a specific output file path."""
    with archive.open("rb") as ifh:
        if sys.version_info >= (3, 14):
            # Use built-in ZstdFile for streaming
            with zstd.ZstdFile(ifh, "rb") as f_in:  # type: ignore
                with out_file.open("wb") as f_out:
                    shutil.copyfileobj(f_in, f_out)
        else:
            dctx = zstd.ZstdDecompressor()
            # Legacy zstandard
            with out_file.open("wb") as f_out:
                dctx.copy_stream(ifh, f_out, write_size=2**20)  # 1 MiB chunks


if __name__ == "__main__":
    p = argparse.ArgumentParser(description="Extract Zstandard (.zst / .tar.zst) files.")
    p.add_argument("archive", help="Path to the .zst or .tar.zst archive to extract.")
    p.add_argument("out_path", help="Directory to extract contents into.")
    args = p.parse_args()
    extract_zstd(args.archive, args.out_path)

test_extract_zst.py

from pathlib import Path
import shutil
import sys
import tarfile

import pytest

from extract_zst import extract_zstd

if sys.version_info >= (3, 14):
    import compression.zstd as zstd  # type: ignore
else:
    import zstandard as zstd  # type: ignore

@pytest.fixture
def repo_root() -> Path:
    return Path(__file__).resolve().parent


def _compress_zstd(source: Path, archive: Path) -> None:
    with source.open("rb") as src, archive.open("wb") as dst:
        if sys.version_info >= (3, 14):
            with zstd.ZstdFile(dst, "wb") as compressed:  # type: ignore[attr-defined]
                shutil.copyfileobj(src, compressed)
        else:
            cctx = zstd.ZstdCompressor()  # type: ignore[attr-defined]
            with cctx.stream_writer(dst) as compressed:
                shutil.copyfileobj(src, compressed)


@pytest.fixture
def sample_paths(tmp_path: Path) -> dict[str, Path]:
    input_dir = tmp_path / "inputs"
    input_dir.mkdir()

    hello_txt = input_dir / "hello.txt"
    goodbye_txt = input_dir / "goodbye.txt"
    hello_zst = input_dir / "hello.txt.zst"
    tar_zst = input_dir / "test.tar.zst"

    hello_txt.write_text("hello\n", encoding="utf-8")
    goodbye_txt.write_text("goodbye\n", encoding="utf-8")

    hello_tar = input_dir / "test.tar"
    with tarfile.open(hello_tar, mode="w") as tar:
        tar.add(hello_txt, arcname="hello.txt")
        tar.add(goodbye_txt, arcname="goodbye.txt")

    _compress_zstd(hello_txt, hello_zst)
    _compress_zstd(hello_tar, tar_zst)

    return {
        "hello_txt": hello_txt,
        "goodbye_txt": goodbye_txt,
        "hello_zst": hello_zst,
        "tar_zst": tar_zst,
    }


@pytest.fixture
def hello_tar(tmp_path: Path, sample_paths: dict[str, Path]) -> Path:
    archive = tmp_path / "hello.tar"
    with tarfile.open(archive, mode="w") as tar:
        tar.add(sample_paths["hello_txt"], arcname="hello.txt")
        tar.add(sample_paths["goodbye_txt"], arcname="goodbye.txt")
    return archive


def test_extract_plain_zst_to_file(tmp_path: Path, sample_paths: dict[str, Path]) -> None:
    extract_zstd(sample_paths["hello_zst"], tmp_path)

    extracted_file = tmp_path / "hello.txt"
    assert extracted_file.exists()
    assert extracted_file.read_bytes() == sample_paths["hello_txt"].read_bytes()


def test_extract_tar_zst_to_directory(tmp_path: Path, sample_paths: dict[str, Path]) -> None:
    extract_zstd(sample_paths["tar_zst"], tmp_path)

    assert (tmp_path / "hello.txt").exists()
    assert (tmp_path / "goodbye.txt").exists()

    # Ensure extraction does not leave an intermediate tar artifact.
    assert not (tmp_path / "test.tar").exists()


def test_extract_plain_tar_mislabeled_input_path(
    tmp_path: Path,
    hello_tar: Path,
) -> None:
    extract_zstd(hello_tar, tmp_path)

    assert (tmp_path / "hello.txt").exists()
    assert (tmp_path / "goodbye.txt").exists()


def test_missing_archive_raises(repo_root: Path, tmp_path: Path) -> None:
    missing = repo_root / "missing-file.zst"

    with pytest.raises(FileNotFoundError):
        extract_zstd(missing, tmp_path)


def test_invalid_zstd_raises_runtime_error(tmp_path: Path) -> None:
    invalid = tmp_path / "invalid.zst"
    invalid.write_bytes(b"not-a-zstd-stream")

    with pytest.raises(zstd.ZstdError):  # type: ignore[attr-defined]
        extract_zstd(invalid, tmp_path)

Thanks

Thanks for this