| 1021 | |
| 1022 | |
| 1023 | class _BytesTarFile(_BufferedWriter): |
| 1024 | def __init__( |
| 1025 | self, |
| 1026 | name: str | None = None, |
| 1027 | mode: Literal["r", "a", "w", "x"] = "r", |
| 1028 | fileobj: ReadBuffer[bytes] | WriteBuffer[bytes] | None = None, |
| 1029 | archive_name: str | None = None, |
| 1030 | **kwargs: Any, |
| 1031 | ) -> None: |
| 1032 | super().__init__() |
| 1033 | self.archive_name = archive_name |
| 1034 | self.name = name |
| 1035 | # error: No overload variant of "open" of "TarFile" matches argument |
| 1036 | # types "str | None", "str", "ReadBuffer[bytes] | WriteBuffer[bytes] | None", |
| 1037 | # "dict[str, Any]" |
| 1038 | # error: Incompatible types in assignment (expression has type "TarFile", |
| 1039 | # base class "_BufferedWriter" defined the type as "BytesIO") |
| 1040 | self.buffer: tarfile.TarFile = tarfile.TarFile.open( # type: ignore[call-overload, assignment] |
| 1041 | name=name, |
| 1042 | mode=self.extend_mode(mode), |
| 1043 | fileobj=fileobj, |
| 1044 | **kwargs, |
| 1045 | ) |
| 1046 | |
| 1047 | def extend_mode(self, mode: str) -> str: |
| 1048 | mode = mode.replace("b", "") |
| 1049 | if mode != "w": |
| 1050 | return mode |
| 1051 | if self.name is not None: |
| 1052 | suffix = Path(self.name).suffix |
| 1053 | if suffix in (".gz", ".xz", ".bz2"): |
| 1054 | mode = f"{mode}:{suffix[1:]}" |
| 1055 | return mode |
| 1056 | |
| 1057 | def infer_filename(self) -> str | None: |
| 1058 | """ |
| 1059 | If an explicit archive_name is not given, we still want the file inside the zip |
| 1060 | file not to be named something.tar, because that causes confusion (GH39465). |
| 1061 | """ |
| 1062 | if self.name is None: |
| 1063 | return None |
| 1064 | |
| 1065 | filename = Path(self.name) |
| 1066 | if filename.suffix == ".tar": |
| 1067 | return filename.with_suffix("").name |
| 1068 | elif filename.suffix in (".tar.gz", ".tar.bz2", ".tar.xz"): |
| 1069 | return filename.with_suffix("").with_suffix("").name |
| 1070 | return filename.name |
| 1071 | |
| 1072 | def write_to_buffer(self) -> None: |
| 1073 | # TarFile needs a non-empty string |
| 1074 | archive_name = self.archive_name or self.infer_filename() or "tar" |
| 1075 | tarinfo = tarfile.TarInfo(name=archive_name) |
| 1076 | tarinfo.size = len(self.getvalue()) |
| 1077 | self.buffer.addfile(tarinfo, self) |
| 1078 | |
| 1079 | |
| 1080 | class _BytesZipFile(_BufferedWriter): |