parallel-compress

asyncio 기반의 병렬 파일 압축/해제 Python 라이브러리입니다.

import asyncio
from parallel_compress import AutoCompress

async def main():
    compressor = AutoCompress(max_concurrent=5)
    results = await compressor.run(
        files=["file1.txt", "file2.csv", "data.json"],
        output="archive.tar.gz",
        format=".tar.gz",
    )
    print(f"✓ {results[0].filename} — {results[0].size_bytes} bytes")

asyncio.run(main())

---

목차

• 특징
• 요구사항
• 설치
• Quick Start
  • 압축
  • 해제
  • 단일 파일 압축
• 지원 포맷
• 고급 사용법
  • 여러 그룹 병렬 압축
  • 커스텀 서브디렉토리명
  • 저수준 핸들러 직접 사용
• 에러 처리
• 플러그인으로 포맷 확장
• 개발
• License

---

특징

⚡ 병렬 처리	asyncio.Semaphore 기반 동시성 제어로 다수의 파일을 빠르게 압축·해제
📦 11개 포맷	아카이브(zip, tar 계열) + 단일 파일 압축(gz, bz2, xz)
🔍 MIME 검증	python-magic으로 해제 전 아카이브 무결성 사전 검증
🛡️ 타입 안전	Pydantic v2 frozen 모델 + 전체 타입 힌트
📊 구조화된 결과	CompressSuccess / CompressFailure 판별 유니온
🚨 계층적 에러	목적별 예외 클래스 + BulkValidationError의 tabulate 테이블 리포트
🔌 플러그인	@register_handler 데코레이터 또는 entry_points로 커스텀 포맷 추가

---

요구사항

• Python ≥ 3.8
• 시스템 의존성: libmagic (python-magic에 필요)

# Ubuntu / Debian
sudo apt install libmagic1

# macOS
brew install libmagic

# Windows — 별도 시스템 설치 불필요
pip install python-magic-bin

---

설치

rye add parallel-compress --git https://github.com/devcomfort/parallel_compress.git
rye sync

브랜치 / 태그 지정

rye add parallel-compress --git https://github.com/devcomfort/parallel_compress.git --branch main  # rye — 브랜치
rye add parallel-compress --git https://github.com/devcomfort/parallel_compress.git --tag v0.1.0   # rye — 태그
rye sync

로컬 (editable)

git clone https://github.com/devcomfort/parallel_compress.git
cd parallel_compress
rye sync             # rye (자동 editable)

---

Quick Start

압축

import asyncio
from parallel_compress import AutoCompress

async def main():
    compressor = AutoCompress(max_concurrent=5)

    results = await compressor.run(
        files=["file1.txt", "file2.csv", "data.json"],
        output="archive.zip",
        # 아카이브: .zip | .tar | .tar.gz (.tgz) | .tar.bz2 (.tbz2) | .tar.xz (.txz)
        # 단일파일: .gz | .bz2 | .xz
        format=".zip",
    )

    for r in results:
        if r.status == "success":
            print(f"✓ {r.filename} ({r.size_bytes} bytes, {r.elapsed_ms}ms)")
        else:
            print(f"✗ {r.filename}: {r.error}")

asyncio.run(main())

해제

import asyncio
from parallel_compress import AutoDecompress

async def main():
    decompressor = AutoDecompress(max_concurrent=5)

    # 각 아카이브 → 아카이브명 기반 서브디렉토리에 추출
    #   archive.zip    → ./extracted/archive/
    #   backup.tar.gz  → ./extracted/backup/
    results = await decompressor.run(
        archives=["archive.zip", "backup.tar.gz"],
        out_dir="./extracted",
    )

    for r in results:
        if r.status == "success":
            print(f"✓ {r.filename} → {r.output_path}")

asyncio.run(main())

단일 파일 압축 (gz / bz2 / xz)

import asyncio
from parallel_compress import AutoCompress

async def main():
    compressor = AutoCompress()

    # 단일 파일 gzip 압축
    results = await compressor.run(
        files=["large_data.csv"],
        output="large_data.csv.gz",
        format=".gz",  # .gz | .bz2 | .xz (단일 파일 전용)
    )

asyncio.run(main())

Note: .gz, .bz2, .xz는 단일 파일만 처리합니다. 여러 파일을 함께 압축하려면 .tar.gz, .tar.bz2, .tar.xz 포맷을 사용하세요.

---

지원 포맷

아카이브 (다중 파일)

포맷	확장자	핸들러	내부 모듈
ZIP	.zip	ZipHandler	zipfile
TAR	.tar	TarHandler	tarfile
TAR + GZIP	.tar.gz, .tgz	TarHandler	tarfile (gz)
TAR + BZIP2	.tar.bz2, .tbz2	TarHandler	tarfile (bz2)
TAR + XZ	.tar.xz, .txz	TarHandler	tarfile (xz)

단일 파일 압축

포맷	확장자	핸들러	내부 모듈
GZIP	.gz	GzipHandler	gzip
BZIP2	.bz2	Bz2Handler	bz2
XZ (LZMA)	.xz	XzHandler	lzma

---

고급 사용법

여러 그룹을 병렬 압축

import asyncio
from parallel_compress import AutoCompress

async def main():
    compressor = AutoCompress(max_concurrent=3)

    # 중첩 리스트 → 각 그룹이 별도 아카이브로
    results = await compressor.run(
        files=[
            ["src/main.py", "src/utils.py"],    # → group_a.tar.gz
            ["docs/api.md", "docs/guide.md"],   # → group_b.tar.gz
        ],
        output=["group_a.tar.gz", "group_b.tar.gz"],
        format=".tar.gz",
    )

asyncio.run(main())

커스텀 서브디렉토리명

results = await decompressor.run(
    archives=["archive.zip", "backup.tar.gz"],
    out_dir="./extracted",
    sub_dir_names={
        "archive.zip": "my_archive",       # → ./extracted/my_archive/
        "backup.tar.gz": "my_backup",      # → ./extracted/my_backup/
    },
)

저수준 핸들러 직접 사용

import asyncio
from pathlib import Path
from parallel_compress.compressor.formats import registry

async def main():
    handler = registry.get_handler(".tar.gz")

    # 압축
    result = await handler.compress(
        files=[Path("file1.txt"), Path("file2.txt")],
        output=Path("output.tar.gz"),
    )

    # 해제
    result = await handler.decompress(
        archive=Path("output.tar.gz"),
        out_dir=Path("./extracted"),
    )

asyncio.run(main())

---

에러 처리

from parallel_compress import (
    AutoCompress,
    BulkValidationError,
    UnsupportedFormatError,
)

async def safe_compress():
    compressor = AutoCompress()
    try:
        await compressor.run(
            files=["missing.txt", "also_missing.txt"],
            output="out.zip",
            format=".zip",
        )
    except BulkValidationError as e:
        # tabulate 테이블로 렌더링된 에러 리포트
        print(e)
        # ╭───┬──────────────────┬────────────────╮
        # │ # │ File Path        │ Reason         │
        # ├───┼──────────────────┼────────────────┤
        # │ 1 │ missing.txt      │ File not found │
        # │ 2 │ also_missing.txt │ File not found │
        # ╰───┴──────────────────┴────────────────╯
    except UnsupportedFormatError as e:
        print(f"미지원 포맷: {e.format}")

예외 계층

CompressError                    ← 최상위 예외
├── UnsupportedFormatError       ← 미지원 포맷 (.rar, .7z 등)
├── CorruptedArchiveError        ← 손상된 아카이브 파일
├── ArchiveNotFoundError         ← 아카이브 파일 없음
├── FileWriteError               ← 파일 쓰기 실패
└── BulkValidationError          ← 배치 유효성 검사 실패 (ExceptionGroup)

---

플러그인으로 포맷 확장

데코레이터 방식

from parallel_compress import BaseCompressHandler, register_handler

@register_handler
class LZ4Handler(BaseCompressHandler):
    supported_extensions = (".lz4",)

    async def compress(self, files, output):
        ...  # lz4 압축 구현

    async def decompress(self, archive, out_dir):
        ...  # lz4 해제 구현

entry_points 방식 (패키지 배포용)

# pyproject.toml
[project.entry-points."parallel_compress.handlers"]
lz4 = "my_package.lz4_handler:LZ4Handler"

---

개발

git clone https://github.com/devcomfort/parallel_compress.git
cd parallel_compress

# ── 의존성 설치 ──
rye sync                                        # rye

# ── 테스트 ──
rye run pytest                                   # rye (커버리지 자동 포함)

# ── 커버리지 리포트 ──
rye run pytest --cov --cov-report=html           # rye → htmlcov/index.html

# ── 린트 & 타입 체크 ──
rye run ruff check src tests                     # rye — 린트
rye run mypy src                                 # rye — 타입 체크

Note: pytest.ini에 --cov=src/parallel_compress --cov-report=term-missing이 기본 설정되어 있어, pytest 실행 시 커버리지가 자동으로 출력됩니다. 최소 커버리지 기준은 80% 입니다.

프로젝트 구조

src/parallel_compress/
├── __init__.py              # Public API re-exports
├── models/
│   ├── types.py             # SupportedFormat, MIME 매핑
│   ├── request.py           # CompressRequest, DecompressRequest
│   └── result.py            # CompressSuccess, CompressFailure
├── errors/
│   ├── compress_errors.py   # 에러 계층
│   └── validation_errors.py # BulkValidationError
├── compressor/
│   ├── base.py              # BaseCompressHandler ABC
│   ├── formats.py           # FormatRegistry, @register_handler
│   ├── auto.py              # AutoCompress, AutoDecompress
│   ├── zip_handler.py       # ZipHandler (.zip)
│   ├── tar_handler.py       # TarHandler (.tar, .tar.gz, ...)
│   ├── gzip_handler.py      # GzipHandler (.gz)
│   ├── bz2_handler.py       # Bz2Handler (.bz2)
│   └── xz_handler.py        # XzHandler (.xz)
└── filesystem/
    └── directory.py          # 출력 디렉토리 / 서브디렉토리 관리

---

License

MIT

Author

DevComfort — GitHub · Email