Archive Utilities

Archive handling for VCollab applications — extract zip/tar.gz archives and stream directories as zip.

Overview

VCollab applications move data in and out as archives: users upload ZIP or TAR.GZ files to be extracted, and request directory contents back as downloadable ZIP archives. This package covers both directions with a memory-vs-speed strategy the caller chooses, and it hardens extraction against hostile input. It provides two categories of functionality:

• Extraction -- Extract ZIP and TAR.GZ archives from any seekable binary stream (BinaryIO) to a target directory, using either in-memory BytesIO (fast, for small files) or temporary file (memory-efficient, for large files) strategies. Includes path traversal protection and configurable size/count limits.
• Streaming -- Generate ZIP archives from directory contents on-the-fly for download responses, with memory-based streaming for small directories and tempfile-based streaming for large directories. Supports file filtering/exclusion.

When to use this package

Use vcti-archive when your application needs to:

• Accept uploaded ZIP or TAR.GZ files and extract them to disk
• Serve directory contents as downloadable ZIP archives
• Stream large directory archives without loading everything into memory
• Choose between memory-efficient and fast extraction strategies
• Protect against malicious archives (path traversal, zip bombs)

Installation

The core package has zero required dependencies — it uses the Python standard library only:

pip install "vcti-archive>=1.0.3"

That gives you everything in the Quick Start: the extractors, create_extractor(), both directory streamers, async wrappers, bomb protection, and path-traversal safety. It works from CLI tools, background workers, or any web framework (Django, Flask, …) that can consume a bytes iterator.

Optional: FastAPI integration

Install the fastapi extra only if you want the streaming_zip_response() helper (see Using with FastAPI):

pip install "vcti-archive[fastapi]>=1.0.3"

Declaring the dependency

# pyproject.toml — core only
dependencies = ["vcti-archive>=1.0.3"]

# …or with the FastAPI helper
dependencies = ["vcti-archive[fastapi]>=1.0.3"]

# requirements.txt
vcti-archive>=1.0.3            # or: vcti-archive[fastapi]>=1.0.3

Quick Start

Extractors accept any seekable BinaryIO (an open file, io.BytesIO, etc.) and the streamers yield plain bytes iterators — no web framework required.

Extract an archive

from pathlib import Path
from vcti.archive import ZipExtractor

with open("archive.zip", "rb") as f:
    extractor = ZipExtractor(f, Path("/target/dir"))
    extractor.extract_using_bytesio()   # fast, reads the archive into memory
    # or, for large archives:
    extractor.extract_using_tempfile()  # memory-efficient, via a temp file

TarGzExtractor has the same interface for .tar.gz / .tgz:

from vcti.archive import TarGzExtractor

with open("archive.tar.gz", "rb") as f:
    TarGzExtractor(f, Path("/target/dir")).extract_using_bytesio()

Select an extractor by filename

When you only have a filename (e.g. an upload), let create_extractor pick the class — it raises UnsupportedArchiveFormat for anything it doesn't recognize:

from vcti.archive import create_extractor, UnsupportedArchiveFormat

try:
    extractor = create_extractor(stream, Path("/target"), filename="upload.tar.gz")
    extractor.extract_using_bytesio()
except UnsupportedArchiveFormat:
    ...  # not a .zip / .tar.gz / .tgz

Guard against malicious archives

Extraction always rejects path-traversal entries (../…). Add optional size and count limits to guard against archive bombs — they are checked against the archive's declared metadata before anything is written to disk:

extractor = ZipExtractor(
    stream, Path("/target"),
    max_total_size=500_000_000,  # 500 MB uncompressed
    max_file_count=10_000,
)
extractor.extract_using_bytesio()  # raises ValueError if a limit is exceeded

Stream a directory as a ZIP

from vcti.archive import DirectoryZipMemoryStreamer

streamer = DirectoryZipMemoryStreamer(Path("/data/project"))
with open("output.zip", "wb") as out:
    for chunk in streamer:
        out.write(chunk)

For directories too large to hold in memory, LargeDirectoryZipStreamer builds the ZIP in a temp file first, then streams it:

from vcti.archive import LargeDirectoryZipStreamer

streamer = LargeDirectoryZipStreamer(
    folder_path=Path("/data/project"),
    archive_name="project.zip",
)
for chunk in streamer.stream():
    out.write(chunk)

Both accept an exclude callback to skip files:

streamer = DirectoryZipMemoryStreamer(
    Path("/data/project"),
    exclude=lambda p: p.name.startswith(".") or p.suffix == ".log",
)

Async extraction

Every extractor has async wrappers that run the (synchronous) stdlib extraction in a worker thread, so they don't block an event loop:

await extractor.async_extract_using_bytesio()
await extractor.async_extract_using_tempfile()

Choosing a streamer

Both streamers produce identical ZIP output. The difference is where the ZIP is assembled:

• DirectoryZipMemoryStreamer — builds the ZIP in a BytesIO buffer, yielding chunks as it goes. Simplest (no temp files, no cleanup), but the buffer stays in memory for the request's duration.
• LargeDirectoryZipStreamer — writes the complete ZIP to a temp file, then streams from disk. Needs cleanup (via on_cleanup or the FastAPI helper) but memory usage stays flat regardless of size.

The right choice depends on your deployment (process memory budget, concurrency, disk speed), not a universal size threshold. Start with DirectoryZipMemoryStreamer and switch to LargeDirectoryZipStreamer if you observe memory pressure under load.

Using with FastAPI

Install the extra (pip install "vcti-archive[fastapi]") to get streaming_zip_response() — a helper that wraps LargeDirectoryZipStreamer in a StreamingResponse with the correct headers and deferred temp-file cleanup via BackgroundTasks. Extraction and the in-memory streamer need no helper: pass UploadFile.file and the streamer straight through.

from pathlib import Path

from fastapi import BackgroundTasks, UploadFile
from fastapi.responses import StreamingResponse

from vcti.archive import (
    DirectoryZipMemoryStreamer,
    LargeDirectoryZipStreamer,
    ZipExtractor,
)
from vcti.archive.fastapi import streaming_zip_response


@app.post("/upload")  # extract an uploaded archive
async def upload(file: UploadFile):
    extractor = ZipExtractor(file.file, Path("/data/uploads"))
    await extractor.async_extract_using_bytesio()
    return {"status": "extracted"}


@app.get("/download")  # small dir — memory streamer, no helper
def download():
    streamer = DirectoryZipMemoryStreamer(Path("/data/project"))
    return StreamingResponse(streamer, media_type="application/zip")


@app.get("/download/large")  # large dir — tempfile streamer + helper
def download_large(background_tasks: BackgroundTasks):
    streamer = LargeDirectoryZipStreamer(
        folder_path=Path("/data/dataset"),
        archive_name="dataset.zip",
    )
    return streaming_zip_response(streamer, background_tasks)

Public API

Dependencies

• Zero required dependencies -- Core functionality uses Python stdlib only (zipfile, tarfile, shutil, tempfile, asyncio).
• Optional: fastapi -- Install with vcti-archive[fastapi] for streaming_zip_response() and FastAPI-specific integration.

Documentation