Skip to content

archive

django_spire.contrib.sync.file.archive

__all__ = ['Archive', 'CollisionStrategy', 'ZipArchive'] module-attribute

Archive

Bases: Protocol

extract

Source code in django_spire/contrib/sync/file/archive/base.py 
10
def extract(self, destination: Path) -> list[Path]: ...

add

Source code in django_spire/contrib/sync/file/archive/base.py 
11
def add(self, path: Path, arcname: str | None = None) -> None: ...

close

Source code in django_spire/contrib/sync/file/archive/base.py 
12
def close(self) -> None: ...

__enter__

Source code in django_spire/contrib/sync/file/archive/base.py 
13
def __enter__(self) -> Self: ...

__exit__

Source code in django_spire/contrib/sync/file/archive/base.py 
14
def __exit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: object) -> None: ...

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