-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathpatch.py
More file actions
368 lines (331 loc) · 12.4 KB
/
Copy pathpatch.py
File metadata and controls
368 lines (331 loc) · 12.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
import copy
import typing
from typing import Any
from diff.diff import Delta
Token = str | int # str for dict keys, int for list indices
class JsonPathError(ValueError):
pass
def _tokenize_json_path(path: str) -> list[Token]:
"""
Tokenize a JSONPath-like string into a list of tokens.
- Dict keys -> strings
- List indices -> integers
Supported syntax:
$.a.b[2]["key.with.dots"][0]
Notes:
- Leading '$' is optional.
- Dot-notation for simple keys.
- Brackets for indices and quoted keys. Quotes can be ' or ".
- Escaping inside quoted keys: backslash escapes the quote and backslash (\", \', \\).
"""
if not isinstance(path, str) or not path:
raise JsonPathError("Path must be a non-empty string.")
i = 0
n = len(path)
tokens: list[Token] = []
# Skip optional leading '$' and optional following '.'
if i < n and path[i] == "$":
i += 1
if i < n and path[i] == ".":
i += 1
def read_simple_key(start: int) -> tuple[str, int]:
j = start
while j < n and path[j] not in ".[":
j += 1
if j == start:
raise JsonPathError(f"Expected key at position {start} in '{path}'")
return path[start:j], j
def read_bracket_key_or_index(start: int) -> tuple[Token, int]:
# start at '[', return (token, new_index_after_'])
j = start + 1
if j >= n:
raise JsonPathError(f"Unclosed '[' at position {start} in '{path}'")
if path[j] in ("'", '"'):
# Quoted key
quote = path[j]
j += 1
buf = []
while j < n:
ch = path[j]
if ch == "\\": # escape sequence
j += 1
if j >= n:
raise JsonPathError("Trailing backslash in quoted key.")
esc = path[j]
if esc in [quote, "\\"]:
buf.append(esc)
else:
# Keep unknown escape as-is (e.g., \n), or handle specially if desired
buf.append(esc)
j += 1
continue
if ch == quote:
j += 1
break
buf.append(ch)
j += 1
else:
raise JsonPathError(
f"Unclosed quoted key starting at position {start} in '{path}'"
)
# Expect closing ']'
if j >= n or path[j] != "]":
raise JsonPathError(
f"Expected ']' after quoted key at position {j} in '{path}'"
)
return "".join(buf), j + 1
# Numeric index
k = j
while k < n and path[k].isdigit():
k += 1
if k == j:
raise JsonPathError(
f"Expected non-negative integer index after '[' at position {start} in '{path}'"
)
if k >= n or path[k] != "]":
raise JsonPathError(f"Expected ']' after index at position {k} in '{path}'")
idx = int(path[j:k])
return idx, k + 1
while i < n:
ch = path[i]
if ch == ".":
i += 1 # skip redundant dots (e.g., '$.a..b' would error on next step)
continue
elif ch == "[":
token, i = read_bracket_key_or_index(i)
tokens.append(token)
else:
key, i = read_simple_key(i)
tokens.append(key)
return tokens
def set_by_json_path(
doc: Any, path: str, value: Any, *, create_missing: bool = True
) -> Any:
"""
Set `value` into `doc` following a JSONPath-like `path`, creating
intermediate dicts/lists if `create_missing` is True.
Mutates `doc` in place and also returns it for convenience.
Raises:
- JsonPathError on path syntax errors.
- TypeError when the path expects a dict/list but finds another type.
- IndexError for negative indices (not supported).
"""
tokens = _tokenize_json_path(path)
if not tokens:
raise JsonPathError("Path resolves to the root; set on '$' is not supported.")
# We'll walk down, creating as needed.
current = doc
parents: list[
tuple[Any, Token]
] = [] # (container, token used to access) if you later want force-replace
for idx, tok in enumerate(tokens):
last = idx == len(tokens) - 1
next_tok = None if last else tokens[idx + 1]
if isinstance(tok, str):
# Expect dict
if not isinstance(current, dict):
raise TypeError(
f"Expected dict at step {idx} for key '{tok}', found {type(current).__name__}"
)
if last:
current[tok] = value
return doc
# create if missing
if tok not in current or current[tok] is None:
if not create_missing:
raise KeyError(
f"Missing key '{tok}' at step {idx} and create_missing=False"
)
current[tok] = [] if isinstance(next_tok, int) else {}
parents.append((current, tok))
current = current[tok]
else:
# list index
index = tok
if index < 0:
raise IndexError(
"Negative indices are not supported in this implementation."
)
if not isinstance(current, list):
raise TypeError(
f"Expected list at step {idx} for index [{index}], found {type(current).__name__}"
)
# extend if needed
if index >= len(current):
if not create_missing:
raise IndexError(
f"Index {index} out of range at step {idx} and create_missing=False"
)
current.extend([None] * (index - len(current) + 1))
if last:
current[index] = value
return doc
if current[index] is None:
if not create_missing:
raise KeyError(
f"Missing element at index {index} (None) and create_missing=False"
)
current[index] = [] if isinstance(next_tok, int) else {}
parents.append((current, index))
current = current[index]
# Should not reach here
return doc
def get_by_json_path(doc: Any, path: str) -> Any:
"""
Retrieve a value from `doc` following the same JSONPath-like syntax.
"""
tokens = _tokenize_json_path(path)
current = doc
for idx, tok in enumerate(tokens):
if isinstance(tok, str):
if not isinstance(current, dict) or tok not in current:
raise KeyError(f"Path segment {tok} not found at step {idx}")
current = current[tok]
else:
if not isinstance(current, list):
raise TypeError(
f"Expected list at step {idx} for index [{tok}], found {type(current).__name__}"
)
if tok < 0 or tok >= len(current):
raise IndexError(f"Index {tok} out of range at step {idx}")
current = current[tok]
return current
def _delete_in_parent(parent: Any, tok: Token, *, remove_from_list: bool) -> None:
"""
Remove child referenced by `tok` from `parent`.
- For dicts: del parent[tok]
- For lists: if remove_from_list=True -> del parent[tok]; else -> parent[tok] = None
"""
if isinstance(tok, str):
if not isinstance(parent, dict):
raise TypeError(
f"Expected dict parent to delete key '{tok}', found {type(parent).__name__}"
)
if tok in parent:
del parent[tok]
else:
# list index
if not isinstance(parent, list):
raise TypeError(
f"Expected list parent to delete index [{tok}], found {type(parent).__name__}"
)
if 0 <= tok < len(parent):
if remove_from_list:
del parent[tok]
else:
parent[tok] = None
def _is_empty_container(obj: Any) -> bool:
"""Return True if obj is an empty dict or empty list."""
return (isinstance(obj, dict) and len(obj) == 0) or (
isinstance(obj, list) and len(obj) == 0
)
def pop_by_json_path(
doc: Any,
path: str,
*,
missing_ok: bool = False,
remove_from_list: bool = False,
prune_empty: bool = False,
) -> Any:
"""
Remove and return the value at `path`. Behaves like delete_by_json_path but returns the removed value.
If the path is missing and missing_ok=True, returns None and leaves `doc` unchanged.
"""
tokens = _tokenize_json_path(path)
if not tokens:
raise JsonPathError("Path resolves to the root; popping '$' is not supported.")
# Traverse to parent
current = doc
parents: list[tuple[Any, Token]] = []
try:
for idx, tok in enumerate(tokens[:-1]):
if isinstance(tok, str):
if (
not isinstance(current, dict)
or tok not in current
or current[tok] is None
):
if missing_ok:
return None
raise KeyError(f"Missing key '{tok}' at step {idx}") # noqa: TRY301
parents.append((current, tok))
current = current[tok]
else:
if not isinstance(current, list) or tok < 0 or tok >= len(current):
if missing_ok:
return None
raise IndexError(f"Index {tok} out of range at step {idx}") # noqa: TRY301
if current[tok] is None:
if missing_ok:
return None
raise KeyError(f"None found at index {tok} at step {idx}") # noqa: TRY301
parents.append((current, tok))
current = current[tok]
except (TypeError, KeyError, IndexError):
if missing_ok:
return None
raise
# Pop at leaf
leaf = tokens[-1]
removed = None
try:
if isinstance(leaf, str):
if not isinstance(current, dict):
if missing_ok:
return None
raise TypeError( # noqa: TRY301
f"Expected dict at leaf for key '{leaf}', found {type(current).__name__}"
)
if leaf not in current:
if missing_ok:
return None
raise KeyError(f"Key '{leaf}' not found at leaf") # noqa: TRY301
removed = current[leaf]
del current[leaf]
else:
if not isinstance(current, list):
if missing_ok:
return None
raise TypeError( # noqa: TRY301
f"Expected list at leaf for index [{leaf}], found {type(current).__name__}"
)
if leaf < 0 or leaf >= len(current):
if missing_ok:
return None
raise IndexError(f"Index {leaf} out of range at leaf") # noqa: TRY301
if remove_from_list:
removed = current[leaf]
del current[leaf]
else:
removed = current[leaf]
current[leaf] = None
except (TypeError, KeyError, IndexError):
if missing_ok:
return None
raise
# Prune if requested
if prune_empty:
child = current
for depth in range(len(parents) - 1, -1, -1):
parent, tok_to_child = parents[depth]
if _is_empty_container(child):
_delete_in_parent(
parent, tok_to_child, remove_from_list=remove_from_list
)
child = parent
else:
break
return removed
def patch(base: dict[str, typing.Any], deltas: list[Delta]) -> dict[str, typing.Any]:
output = copy.deepcopy(base)
for op in deltas:
if op.operation == "deleted":
pop_by_json_path(output, op.path, prune_empty=True, remove_from_list=True)
continue
# we use set for both modify and add actions
set_by_json_path(
output,
op.path,
op.new_value,
)
return output