|
5 | 5 | import sys |
6 | 6 | from abc import ABC, abstractmethod |
7 | 7 | from functools import cached_property |
8 | | -from typing import TYPE_CHECKING |
| 8 | +from typing import TYPE_CHECKING, Literal |
9 | 9 |
|
10 | 10 | from hatch.config.constants import PythonEnvVars |
11 | 11 | from hatch.errors import PythonDistributionResolutionError, PythonDistributionUnknownError |
@@ -55,43 +55,36 @@ def archive_name(self) -> str: |
55 | 55 | return self.source.rsplit("/", 1)[-1] |
56 | 56 |
|
57 | 57 | def unpack(self, archive: Path, directory: Path) -> None: |
| 58 | + # zip |
58 | 59 | if self.source.endswith(".zip"): |
59 | 60 | import zipfile |
60 | 61 |
|
61 | 62 | with zipfile.ZipFile(archive, "r") as zf: |
62 | 63 | zf.extractall(directory) |
63 | | - elif self.source.endswith((".tar.gz", ".tgz")): |
64 | | - import tarfile |
| 64 | + return |
65 | 65 |
|
66 | | - with tarfile.open(archive, "r:gz") as tf: |
67 | | - if sys.version_info[:2] >= (3, 12): |
68 | | - tf.extractall(directory, filter="data") |
69 | | - else: |
70 | | - tf.extractall(directory) # noqa: S202 |
71 | | - elif self.source.endswith((".tar.bz2", ".bz2")): |
| 66 | + # tar |
| 67 | + if sys.version_info >= (3, 14): |
72 | 68 | import tarfile |
| 69 | + else: |
| 70 | + # for zstd support (introduced in Python 3.14) |
| 71 | + # and filter kwarg (introduced in Python 3.12) |
| 72 | + from backports.zstd import tarfile |
73 | 73 |
|
74 | | - with tarfile.open(archive, "r:bz2") as tf: |
75 | | - if sys.version_info[:2] >= (3, 12): |
76 | | - tf.extractall(directory, filter="data") |
77 | | - else: |
78 | | - tf.extractall(directory) # noqa: S202 |
| 74 | + mode: Literal["r:gz", "r:bz2", "r:zst"] |
| 75 | + if self.source.endswith((".tar.gz", ".tgz")): |
| 76 | + mode = "r:gz" |
| 77 | + elif self.source.endswith((".tar.bz2", ".bz2")): |
| 78 | + mode = "r:bz2" |
79 | 79 | elif self.source.endswith((".tar.zst", ".tar.zstd")): |
80 | | - import tarfile |
81 | | - |
82 | | - import zstandard |
83 | | - |
84 | | - with open(archive, "rb") as ifh: |
85 | | - dctx = zstandard.ZstdDecompressor() |
86 | | - with dctx.stream_reader(ifh) as reader, tarfile.open(mode="r|", fileobj=reader) as tf: |
87 | | - if sys.version_info[:2] >= (3, 12): |
88 | | - tf.extractall(directory, filter="data") |
89 | | - else: |
90 | | - tf.extractall(directory) # noqa: S202 |
| 80 | + mode = "r:zst" |
91 | 81 | else: |
92 | 82 | message = f"Unknown archive type: {archive}" |
93 | 83 | raise ValueError(message) |
94 | 84 |
|
| 85 | + with tarfile.open(archive, mode) as tf: |
| 86 | + tf.extractall(directory, filter="data") |
| 87 | + |
95 | 88 | @property |
96 | 89 | @abstractmethod |
97 | 90 | def version(self) -> Version: |
|
0 commit comments