|
19 | 19 | import io |
20 | 20 | import os |
21 | 21 | import pathlib |
22 | | -from typing import overload |
23 | | -from typing_extensions import TypeGuard |
| 22 | +from typing import Sequence, cast, overload |
| 23 | +from typing_extensions import TypeVar, TypeGuard |
24 | 24 |
|
25 | 25 | import anyio |
26 | 26 |
|
|
33 | 33 | HttpxFileContent, |
34 | 34 | HttpxRequestFiles, |
35 | 35 | ) |
36 | | -from ._utils import is_tuple_t, is_mapping_t, is_sequence_t |
| 36 | +from ._utils import is_list, is_mapping, is_tuple_t, is_mapping_t, is_sequence_t |
| 37 | + |
| 38 | +_T = TypeVar("_T") |
37 | 39 |
|
38 | 40 |
|
39 | 41 | def is_base64_file_input(obj: object) -> TypeGuard[Base64FileInput]: |
@@ -137,3 +139,51 @@ async def async_read_file_content(file: FileContent) -> HttpxFileContent: |
137 | 139 | return await anyio.Path(file).read_bytes() |
138 | 140 |
|
139 | 141 | return file |
| 142 | + |
| 143 | + |
| 144 | +def deepcopy_with_paths(item: _T, paths: Sequence[Sequence[str]]) -> _T: |
| 145 | + """Copy only the containers along the given paths. |
| 146 | +
|
| 147 | + Used to guard against mutation by extract_files without copying the entire structure. |
| 148 | + Only dicts and lists that lie on a path are copied; everything else |
| 149 | + is returned by reference. |
| 150 | +
|
| 151 | + For example, given paths=[["foo", "files", "file"]] and the structure: |
| 152 | + { |
| 153 | + "foo": { |
| 154 | + "bar": {"baz": {}}, |
| 155 | + "files": {"file": <content>} |
| 156 | + } |
| 157 | + } |
| 158 | + The root dict, "foo", and "files" are copied (they lie on the path). |
| 159 | + "bar" and "baz" are returned by reference (off the path). |
| 160 | + """ |
| 161 | + return _deepcopy_with_paths(item, paths, 0) |
| 162 | + |
| 163 | + |
| 164 | +def _deepcopy_with_paths(item: _T, paths: Sequence[Sequence[str]], index: int) -> _T: |
| 165 | + if not paths: |
| 166 | + return item |
| 167 | + if is_mapping(item): |
| 168 | + key_to_paths: dict[str, list[Sequence[str]]] = {} |
| 169 | + for path in paths: |
| 170 | + if index < len(path): |
| 171 | + key_to_paths.setdefault(path[index], []).append(path) |
| 172 | + |
| 173 | + # if no path continues through this mapping, it won't be mutated and copying it is redundant |
| 174 | + if not key_to_paths: |
| 175 | + return item |
| 176 | + |
| 177 | + result = dict(item) |
| 178 | + for key, subpaths in key_to_paths.items(): |
| 179 | + if key in result: |
| 180 | + result[key] = _deepcopy_with_paths(result[key], subpaths, index + 1) |
| 181 | + return cast(_T, result) |
| 182 | + if is_list(item): |
| 183 | + array_paths = [path for path in paths if index < len(path) and path[index] == "<array>"] |
| 184 | + |
| 185 | + # if no path expects a list here, nothing will be mutated inside it - return by reference |
| 186 | + if not array_paths: |
| 187 | + return cast(_T, item) |
| 188 | + return cast(_T, [_deepcopy_with_paths(entry, array_paths, index + 1) for entry in item]) |
| 189 | + return item |
0 commit comments