3755347159

Build Type

push

github

Committed by Tristan Sweeney

Commit Message

Add JSONPath style verbs to leverage pointers

Run Details

123 of 123 new or added lines in 2 files covered. (100.0%)

203 of 222 relevant lines covered (91.44%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.05

/src/fast_json_pointer/resolver.py

from dataclasses import dataclass, field
from typing import *

from .exceptions import EndOfArrayException, ResolutionException
from .pointer import JsonPointer, RelativeJsonPointer

JsonType = dict[str, "JsonType"] | list["JsonType"] | str | bool | int | float | None


@dataclass
class JsonRef:
    doc: JsonType
    pointer: JsonPointer


def _resolve_ref(doc: JsonType, part: str) -> JsonType:
    match doc:
        case dict():
            if part not in doc:
                raise ResolutionException(f"Key '{part}' not in JSON object")

            return doc[part]

        case list():
            if part == "-":
                raise EndOfArrayException("Hit '-' (end of array) token")

            part_idx = int(part)
            if part_idx >= len(doc):
                raise ResolutionException(f"Index '{part_idx}' not in JSON array")

            return doc[part_idx]

        case _:
            raise ResolutionException(f"Unnvaigable doc type '{type(part)}'")


def _resolve(doc: JsonType, pointer: JsonPointer, *, base_pointer: JsonPointer | None = None) -> list[JsonRef]:
    doc_pointer = JsonPointer([]) if base_pointer is None else base_pointer
    doc_refs = [JsonRef(doc, doc_pointer)]

    for idx, part in enumerate(pointer.parts):
        try:
            doc = _resolve_ref(doc, part)
        except Exception as e:
            raise ResolutionException(
                "Error resolving json pointer",
                doc_refs=doc_refs,
                remaining=pointer.parts[idx:],
            ) from e

        doc_pointer = JsonPointer([*doc_pointer.parts, part])
        doc_refs.append(JsonRef(doc, doc_pointer))

    return doc_refs



def resolve(
    doc: JsonType, pointer: JsonPointer, *, rel: RelativeJsonPointer | None = None
) -> list[JsonRef]:
    doc_refs = _resolve(doc, pointer)

    if rel:
        if rel.offset > 0:
            doc_refs = doc_refs[: -rel.offset]

        last_ref = doc_refs[-1]
        
        if rel.is_index_ref:
            return doc_refs + [JsonRef(last_ref.pointer.parts[-1], last_ref.pointer)]

        new_refs = _resolve(last_ref.doc, rel.pointer, base_pointer=last_ref.pointer)
        doc_refs.extend(new_refs)

    return doc_refs



def get(
    doc: JsonType,
    pointer: str | JsonPointer,
    *,
    rel: str | RelativeJsonPointer | None = None,
) -> JsonType:
    """

    >>> get({}, "")
    {}
    >>> get({'x': 5}, "/x")
    5
    >>> get({'x': {'': 3}}, "/x/")
    3
    >>> get({'x': {'': 3, 'z': 12}}, "/x/", rel="1/z")
    12
    >>> get({'x': {'': 3}, 'z': 12}, "/x/", rel="1#")
    'x'
    >>> get([{'x': {'': 3}}, 4], "/0/x", rel="1#")
    '0'

    Trying to get fields that don't exist is a bad idea...

    >>> get([{'x': {'': 3}}, 4], "/0/x", rel="0//does-not-exist")
    Traceback (most recent call last):
    fast_json_pointer.exceptions.ResolutionException: ...
    >>> get([{'x': {'': 3}}, 4], "/3")
    Traceback (most recent call last):
    fast_json_pointer.exceptions.ResolutionException: ...
    >>> get([{'x': {'': 3}}, 4], "/0/z")
    Traceback (most recent call last):
    fast_json_pointer.exceptions.ResolutionException: ...
    """
    match pointer:
        case str():
            pointer = JsonPointer.parse(pointer)

    match rel:
        case str():
            rel = RelativeJsonPointer.parse(rel)

    doc_refs = resolve(doc, pointer, rel=rel)

    return doc_refs[-1].doc


def _set_ref(doc: JsonType, part: str, value: JsonType) -> None:
    match doc:
        case dict():
            doc[part] = value
        case list():
            part_idx = int(part)
            doc[part_idx] = value
        case _:
            raise RuntimeError(f"Unnavigable type {type(doc)}")


def add(
    doc: JsonType,
    pointer: str | JsonPointer,
    value: JsonType,
    *,
    rel: str | RelativeJsonPointer | None = None,
) -> None:
    """
    >>> obj = {}
    >>> add(obj, "/x", 2)
    >>> obj
    {'x': 2}

    >>> obj = {'x': 2}
    >>> add(obj, "", 'foo', rel="0/y")
    >>> obj
    {'x': 2, 'y': 'foo'}

    >>> obj = {'x': 2}
    >>> add(obj, "/x", 'foo', rel="1/x")
    >>> obj
    {'x': 'foo'}

    >>> obj = {'x': 2}
    >>> add(obj, "/x", 'foo', rel="1/y")
    >>> obj
    {'x': 2, 'y': 'foo'}
    """

    match pointer:
        case str():
            pointer = JsonPointer.parse(pointer)

    match rel:
        case str():
            rel = RelativeJsonPointer.parse(rel)

    if rel and rel.is_index_ref:
        raise RuntimeError()

    try:
        doc_refs = resolve(doc, pointer, rel=rel)
        parent = doc_refs[-2].doc
        part = rel.pointer.parts[-1] if rel is not None else pointer.parts[-1]
    except ResolutionException as e:
        if len(e.remaining) > 1:
            raise

        parent = e.doc_refs[-1].doc
        part = e.remaining[0]

    _set_ref(parent, part, value)


def remove(doc, pointer: str | JsonPointer,  *, rel: str | RelativeJsonPointer | None = None) -> None:
    '''
    >>> obj = {'x': 2}
    >>> remove(obj, "/x")
    >>> obj
    {}
    '''
    match pointer:
        case str():
            pointer = JsonPointer.parse(pointer)

    match rel:
        case str():
            rel = RelativeJsonPointer.parse(rel)

    doc_refs = resolve(doc, pointer, rel=rel)
    parent = doc_refs[-2]
    last_ref = doc_refs[-1]

    part = last_ref.pointer.parts[-1]

    match parent.doc:
        case dict():
            del parent.doc[part]
        case list():
            del parent.doc[int(part)]
        case _:
            raise RuntimeError()
    

def replace(doc, pointer: str | JsonPointer, value: JsonType, *, rel: str | RelativeJsonPointer | None = None) -> None:
    '''
    >>> obj = {'x': 2}
    >>> replace(obj, "/x", ['foo'])
    >>> obj
    {'x': ['foo']}
    '''
    
    match pointer:
        case str():
            pointer = JsonPointer.parse(pointer)

    match rel:
        case str():
            rel = RelativeJsonPointer.parse(rel)

    doc_refs = resolve(doc, pointer, rel=rel)
    parent = doc_refs[-2]
    last_ref = doc_refs[-1]

    part = last_ref.pointer.parts[-1]

    match parent.doc:
        case dict():
            parent.doc[part] = value
        case list():
            parent.doc.insert(int(part), value)
        case _:
            raise RuntimeError()


def move(doc, from_: str | JsonPointer, pointer: str | JsonPointer, *, rel: str | RelativeJsonPointer | None = None, from_rel: str | RelativeJsonPointer | None = None) -> None:
    '''
    >>> obj = {'x': 2}
    >>> move(obj, "/x", "/y")
    >>> obj
    {'y': 2}
    '''
    
    obj = get(doc, from_, rel=from_rel)
    remove(doc, from_, rel=from_rel)
    add(doc, pointer, obj, rel=rel)


def copy(doc, from_: str | JsonPointer, pointer: str | JsonPointer, *, rel: str | RelativeJsonPointer | None = None, from_rel: str | RelativeJsonPointer | None = None) -> None:
    '''
    >>> obj = {'x': 2}
    >>> copy(obj, "/x", "/y")
    >>> obj
    {'x': 2, 'y': 2}
    '''
    obj = get(doc, from_, rel=from_rel)
    add(doc, pointer, obj, rel=rel)


def test(doc, pointer: str | JsonPointer, value: JsonType, *, rel: str | RelativeJsonPointer | None = None) -> bool:
    '''
    >>> obj = {'x': 2}
    >>> test(obj, "/x", 2)
    True
    '''
    obj = get(doc, pointer, rel=rel)
    return obj == value

1	from dataclasses import dataclass, field	1✔
2	from typing import *	1✔
3
4	from .exceptions import EndOfArrayException, ResolutionException	1✔
5	from .pointer import JsonPointer, RelativeJsonPointer	1✔
6
7	JsonType = dict[str, "JsonType"] \| list["JsonType"] \| str \| bool \| int \| float \| None	1✔
8
9
10	@dataclass	1✔
11	class JsonRef:	1✔
12	doc: JsonType	1✔
13	pointer: JsonPointer	1✔
14
15
16	def _resolve_ref(doc: JsonType, part: str) -> JsonType:	1✔
17	match doc:	1✔
18	case dict():	1✔
19	if part not in doc:	1✔
20	raise ResolutionException(f"Key '{part}' not in JSON object")	1✔
21
22	return doc[part]	1✔
23
24	case list():	1✔
25	if part == "-":	1✔
26	raise EndOfArrayException("Hit '-' (end of array) token")	×
27
28	part_idx = int(part)	1✔
29	if part_idx >= len(doc):	1✔
30	raise ResolutionException(f"Index '{part_idx}' not in JSON array")	1✔
31
32	return doc[part_idx]	1✔
33
34	case _:	1✔
35	raise ResolutionException(f"Unnvaigable doc type '{type(part)}'")	1✔
36
37
38	def _resolve(doc: JsonType, pointer: JsonPointer, *, base_pointer: JsonPointer \| None = None) -> list[JsonRef]:	1✔
39	doc_pointer = JsonPointer([]) if base_pointer is None else base_pointer	1✔
40	doc_refs = [JsonRef(doc, doc_pointer)]	1✔
41
42	for idx, part in enumerate(pointer.parts):	1✔
43	try:	1✔
44	doc = _resolve_ref(doc, part)	1✔
45	except Exception as e:	1✔
46	raise ResolutionException(	1✔
47	"Error resolving json pointer",
48	doc_refs=doc_refs,
49	remaining=pointer.parts[idx:],
50	) from e
51
52	doc_pointer = JsonPointer([*doc_pointer.parts, part])	1✔
53	doc_refs.append(JsonRef(doc, doc_pointer))	1✔
54
55	return doc_refs	1✔
56
57
58
59	def resolve(	1✔
60	doc: JsonType, pointer: JsonPointer, *, rel: RelativeJsonPointer \| None = None
61	) -> list[JsonRef]:
62	doc_refs = _resolve(doc, pointer)	1✔
63
64	if rel:	1✔
65	if rel.offset > 0:	1✔
66	doc_refs = doc_refs[: -rel.offset]	1✔
67
68	last_ref = doc_refs[-1]	1✔
69
70	if rel.is_index_ref:	1✔
71	return doc_refs + [JsonRef(last_ref.pointer.parts[-1], last_ref.pointer)]	1✔
72
73	new_refs = _resolve(last_ref.doc, rel.pointer, base_pointer=last_ref.pointer)	1✔
74	doc_refs.extend(new_refs)	1✔
75
76	return doc_refs	1✔
77
78
79
80	def get(	1✔
81	doc: JsonType,
82	pointer: str \| JsonPointer,
83	*,
84	rel: str \| RelativeJsonPointer \| None = None,
85	) -> JsonType:
86	"""
87
88	>>> get({}, "")
89	{}
90	>>> get({'x': 5}, "/x")
91	5
92	>>> get({'x': {'': 3}}, "/x/")
93	3
94	>>> get({'x': {'': 3, 'z': 12}}, "/x/", rel="1/z")
95	12
96	>>> get({'x': {'': 3}, 'z': 12}, "/x/", rel="1#")
97	'x'
98	>>> get([{'x': {'': 3}}, 4], "/0/x", rel="1#")
99	'0'
100
101	Trying to get fields that don't exist is a bad idea...
102
103	>>> get([{'x': {'': 3}}, 4], "/0/x", rel="0//does-not-exist")
104	Traceback (most recent call last):
105	fast_json_pointer.exceptions.ResolutionException: ...
106	>>> get([{'x': {'': 3}}, 4], "/3")
107	Traceback (most recent call last):
108	fast_json_pointer.exceptions.ResolutionException: ...
109	>>> get([{'x': {'': 3}}, 4], "/0/z")
110	Traceback (most recent call last):
111	fast_json_pointer.exceptions.ResolutionException: ...
112	"""
113	match pointer:	1✔
114	case str():	1✔
115	pointer = JsonPointer.parse(pointer)	1✔
116
117	match rel:	1✔
118	case str():	1✔
119	rel = RelativeJsonPointer.parse(rel)	1✔
120
121	doc_refs = resolve(doc, pointer, rel=rel)	1✔
122
123	return doc_refs[-1].doc	1✔
124
125
126	def _set_ref(doc: JsonType, part: str, value: JsonType) -> None:	1✔
127	match doc:	1✔
128	case dict():	1✔
129	doc[part] = value	1✔
130	case list():	×
131	part_idx = int(part)	×
132	doc[part_idx] = value	×
133	case _:	×
134	raise RuntimeError(f"Unnavigable type {type(doc)}")	×
135
136
137	def add(	1✔
138	doc: JsonType,
139	pointer: str \| JsonPointer,
140	value: JsonType,
141	*,
142	rel: str \| RelativeJsonPointer \| None = None,
143	) -> None:
144	"""
145	>>> obj = {}
146	>>> add(obj, "/x", 2)
147	>>> obj
148	{'x': 2}
149
150	>>> obj = {'x': 2}
151	>>> add(obj, "", 'foo', rel="0/y")
152	>>> obj
153	{'x': 2, 'y': 'foo'}
154
155	>>> obj = {'x': 2}
156	>>> add(obj, "/x", 'foo', rel="1/x")
157	>>> obj
158	{'x': 'foo'}
159
160	>>> obj = {'x': 2}
161	>>> add(obj, "/x", 'foo', rel="1/y")
162	>>> obj
163	{'x': 2, 'y': 'foo'}
164	"""
165
166	match pointer:	1✔
167	case str():	1✔
168	pointer = JsonPointer.parse(pointer)	1✔
169
170	match rel:	1✔
171	case str():	1✔
172	rel = RelativeJsonPointer.parse(rel)	1✔
173
174	if rel and rel.is_index_ref:	1✔
175	raise RuntimeError()	×
176
177	try:	1✔
178	doc_refs = resolve(doc, pointer, rel=rel)	1✔
179	parent = doc_refs[-2].doc	1✔
180	part = rel.pointer.parts[-1] if rel is not None else pointer.parts[-1]	1✔
181	except ResolutionException as e:	1✔
182	if len(e.remaining) > 1:	1✔
183	raise	×
184
185	parent = e.doc_refs[-1].doc	1✔
186	part = e.remaining[0]	1✔
187
188	_set_ref(parent, part, value)	1✔
189
190
191	def remove(doc, pointer: str \| JsonPointer, *, rel: str \| RelativeJsonPointer \| None = None) -> None:	1✔
192	'''
193	>>> obj = {'x': 2}
194	>>> remove(obj, "/x")
195	>>> obj
196	{}
197	'''
198	match pointer:	1✔
199	case str():	1✔
200	pointer = JsonPointer.parse(pointer)	1✔
201
202	match rel:	1✔
203	case str():	1✔
204	rel = RelativeJsonPointer.parse(rel)	×
205
206	doc_refs = resolve(doc, pointer, rel=rel)	1✔
207	parent = doc_refs[-2]	1✔
208	last_ref = doc_refs[-1]	1✔
209
210	part = last_ref.pointer.parts[-1]	1✔
211
212	match parent.doc:	1✔
213	case dict():	1✔
214	del parent.doc[part]	1✔
215	case list():	×
216	del parent.doc[int(part)]	×
217	case _:	×
218	raise RuntimeError()	×
219
220
221	def replace(doc, pointer: str \| JsonPointer, value: JsonType, *, rel: str \| RelativeJsonPointer \| None = None) -> None:	1✔
222	'''
223	>>> obj = {'x': 2}
224	>>> replace(obj, "/x", ['foo'])
225	>>> obj
226	{'x': ['foo']}
227	'''
228
229	match pointer:	1✔
230	case str():	1✔
231	pointer = JsonPointer.parse(pointer)	1✔
232
233	match rel:	1✔
234	case str():	1✔
235	rel = RelativeJsonPointer.parse(rel)	×
236
237	doc_refs = resolve(doc, pointer, rel=rel)	1✔
238	parent = doc_refs[-2]	1✔
239	last_ref = doc_refs[-1]	1✔
240
241	part = last_ref.pointer.parts[-1]	1✔
242
243	match parent.doc:	1✔
244	case dict():	1✔
245	parent.doc[part] = value	1✔
246	case list():	×
247	parent.doc.insert(int(part), value)	×
248	case _:	×
249	raise RuntimeError()	×
250
251
252	def move(doc, from_: str \| JsonPointer, pointer: str \| JsonPointer, *, rel: str \| RelativeJsonPointer \| None = None, from_rel: str \| RelativeJsonPointer \| None = None) -> None:	1✔
253	'''
254	>>> obj = {'x': 2}
255	>>> move(obj, "/x", "/y")
256	>>> obj
257	{'y': 2}
258	'''
259
260	obj = get(doc, from_, rel=from_rel)	1✔
261	remove(doc, from_, rel=from_rel)	1✔
262	add(doc, pointer, obj, rel=rel)	1✔
263
264
265	def copy(doc, from_: str \| JsonPointer, pointer: str \| JsonPointer, *, rel: str \| RelativeJsonPointer \| None = None, from_rel: str \| RelativeJsonPointer \| None = None) -> None:	1✔
266	'''
267	>>> obj = {'x': 2}
268	>>> copy(obj, "/x", "/y")
269	>>> obj
270	{'x': 2, 'y': 2}
271	'''
272	obj = get(doc, from_, rel=from_rel)	1✔
273	add(doc, pointer, obj, rel=rel)	1✔
274
275
276	def test(doc, pointer: str \| JsonPointer, value: JsonType, *, rel: str \| RelativeJsonPointer \| None = None) -> bool:	1✔
277	'''
278	>>> obj = {'x': 2}
279	>>> test(obj, "/x", 2)
280	True
281	'''
282	obj = get(doc, pointer, rel=rel)	1✔
283	return obj == value	1✔

SlowAPI / fast-json-pointer / 3755347159

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous