Skip to content

zip

django_spire.contrib.sync.file.archive.zip

logger = logging.getLogger(__name__) module-attribute

CollisionStrategy

Bases: Enum

RAISE = 'raise' class-attribute instance-attribute

SKIP = 'skip' class-attribute instance-attribute

OVERWRITE = 'overwrite' class-attribute instance-attribute

ZipArchive

Bases: Archive

Source code in django_spire/contrib/sync/file/archive/zip.py 
23
24
25
26
27
28
29
30
31
def __init__(
    self,
    encoding: str = 'utf-8',
    flatten: bool = True,
    collision: CollisionStrategy = CollisionStrategy.RAISE,
) -> None:
    self._encoding = encoding
    self._flatten = flatten
    self._collision = collision

extract

Source code in django_spire/contrib/sync/file/archive/zip.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def extract(self, archive_path: Path, target_dir: Path) -> list[Path]:
    target_dir.mkdir(parents=True, exist_ok=True)

    seen: set[str] = set()
    extracted: list[Path] = []

    with zipfile.ZipFile(archive_path, 'r') as zf:
        for info in zf.infolist():
            if info.is_dir():
                continue

            target = self._resolve_target(info, target_dir)

            if not target.name:
                continue

            target_key = str(target)

            if target_key in seen:
                if self._collision is CollisionStrategy.RAISE:
                    message = (
                        f'Duplicate target {target} in archive '
                        f'{archive_path} (from {info.filename!r})'
                    )

                    raise FileSyncArchiveError(message)

                if self._collision is CollisionStrategy.SKIP:
                    logger.debug(
                        'Skipping duplicate %s from %s',
                        target,
                        info.filename,
                    )
                    continue

            seen.add(target_key)

            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(zf.read(info.filename))
            extracted.append(target)

    logger.info(
        'Extracted %d files from %s to %s',
        len(extracted),
        archive_path,
        target_dir,
    )

    return extracted