|
3 | 3 | import io |
4 | 4 | import os |
5 | 5 | import pathlib |
6 | | -from typing import overload |
7 | | -from typing_extensions import TypeGuard |
| 6 | +from typing import Sequence, cast, overload |
| 7 | +from typing_extensions import TypeVar, TypeGuard |
8 | 8 |
|
9 | 9 | import anyio |
10 | 10 |
|
|
17 | 17 | HttpxFileContent, |
18 | 18 | HttpxRequestFiles, |
19 | 19 | ) |
20 | | -from ._utils import is_tuple_t, is_mapping_t, is_sequence_t |
| 20 | +from ._utils import is_list, is_mapping, is_tuple_t, is_mapping_t, is_sequence_t |
| 21 | + |
| 22 | +_T = TypeVar("_T") |
21 | 23 |
|
22 | 24 |
|
23 | 25 | def is_base64_file_input(obj: object) -> TypeGuard[Base64FileInput]: |
@@ -121,3 +123,51 @@ async def async_read_file_content(file: FileContent) -> HttpxFileContent: |
121 | 123 | return await anyio.Path(file).read_bytes() |
122 | 124 |
|
123 | 125 | return file |
| 126 | + |
| 127 | + |
| 128 | +def deepcopy_with_paths(item: _T, paths: Sequence[Sequence[str]]) -> _T: |
| 129 | + """Copy only the containers along the given paths. |
| 130 | +
|
| 131 | + Used to guard against mutation by extract_files without copying the entire structure. |
| 132 | + Only dicts and lists that lie on a path are copied; everything else |
| 133 | + is returned by reference. |
| 134 | +
|
| 135 | + For example, given paths=[["foo", "files", "file"]] and the structure: |
| 136 | + { |
| 137 | + "foo": { |
| 138 | + "bar": {"baz": {}}, |
| 139 | + "files": {"file": <content>} |
| 140 | + } |
| 141 | + } |
| 142 | + The root dict, "foo", and "files" are copied (they lie on the path). |
| 143 | + "bar" and "baz" are returned by reference (off the path). |
| 144 | + """ |
| 145 | + return _deepcopy_with_paths(item, paths, 0) |
| 146 | + |
| 147 | + |
| 148 | +def _deepcopy_with_paths(item: _T, paths: Sequence[Sequence[str]], index: int) -> _T: |
| 149 | + if not paths: |
| 150 | + return item |
| 151 | + if is_mapping(item): |
| 152 | + key_to_paths: dict[str, list[Sequence[str]]] = {} |
| 153 | + for path in paths: |
| 154 | + if index < len(path): |
| 155 | + key_to_paths.setdefault(path[index], []).append(path) |
| 156 | + |
| 157 | + # if no path continues through this mapping, it won't be mutated and copying it is redundant |
| 158 | + if not key_to_paths: |
| 159 | + return item |
| 160 | + |
| 161 | + result = dict(item) |
| 162 | + for key, subpaths in key_to_paths.items(): |
| 163 | + if key in result: |
| 164 | + result[key] = _deepcopy_with_paths(result[key], subpaths, index + 1) |
| 165 | + return cast(_T, result) |
| 166 | + if is_list(item): |
| 167 | + array_paths = [path for path in paths if index < len(path) and path[index] == "<array>"] |
| 168 | + |
| 169 | + # if no path expects a list here, nothing will be mutated inside it - return by reference |
| 170 | + if not array_paths: |
| 171 | + return cast(_T, item) |
| 172 | + return cast(_T, [_deepcopy_with_paths(entry, array_paths, index + 1) for entry in item]) |
| 173 | + return item |
0 commit comments