Metadata-Version: 2.4
Name: vcti-archive
Version: 1.0.3
Summary: Archive handling for VCollab applications — extract zip/tar.gz archives and stream directories as zip
Author: Visual Collaboration Technologies Inc.
License-Expression: LicenseRef-Proprietary
Project-URL: Repository, https://github.com/vcollab/vcti-python-archive
Project-URL: Changelog, https://github.com/vcollab/vcti-python-archive/blob/main/CHANGELOG.md
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Python: <3.15,>=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: fastapi
Requires-Dist: fastapi; extra == "fastapi"
Provides-Extra: test
Requires-Dist: pytest; extra == "test"
Requires-Dist: pytest-cov; extra == "test"
Requires-Dist: fastapi; extra == "test"
Provides-Extra: lint
Requires-Dist: ruff; extra == "lint"
Provides-Extra: typecheck
Requires-Dist: mypy; extra == "typecheck"
Dynamic: license-file

# 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:

```bash
pip install "vcti-archive>=1.0.3"
```

That gives you everything in the [Quick Start](#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](#using-with-fastapi)):

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

### Declaring the dependency

```toml
# 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

```python
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`:

```python
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:

```python
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:

```python
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

```python
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:

```python
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:

```python
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:

```python
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.

```python
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

| Class / Function | Purpose |
|------------------|---------|
| `ArchiveExtractor` | ABC base class for archive extractors (BytesIO and tempfile strategies) |
| `ZipExtractor` | Extract ZIP archives with path traversal and bomb protection |
| `TarGzExtractor` | Extract TAR.GZ archives with `filter="data"` security |
| `create_extractor()` | Select `ZipExtractor`/`TarGzExtractor` by filename extension |
| `DirectoryZipMemoryStreamer` | Stream directory as ZIP using in-memory buffer (reusable) |
| `LargeDirectoryZipStreamer` | Stream directory as ZIP using temporary file |
| `UnsupportedArchiveFormat` | Exception for unsupported archive formats |
| `streaming_zip_response()` | FastAPI helper (optional, requires `vcti-archive[fastapi]`) |

---

## 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

| If you want to… | Read |
|---|---|
| Get started using the package | Quick Start above |
| See practical, real-world usage | [docs/patterns.md](docs/patterns.md) |
| Understand the architecture and design decisions | [docs/design.md](docs/design.md) |
| Navigate and understand the source | [docs/source-guide.md](docs/source-guide.md) |
| Add a new archive format | [docs/extending.md](docs/extending.md) |
| Look up a specific function or type | [docs/api.md](docs/api.md) |
| Review the security model | [SECURITY.md](SECURITY.md) |
