3798344209

Build Type

push

github

Committed by Tristan Sweeney

Commit Message

Clean up a bit

Run Details

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

285 of 294 relevant lines covered (96.94%)

0.97 hits per line

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

100.0

/src/fast_json_pointer/resolver.py

from dataclasses import dataclass
from typing import *

from .exceptions import (
    CompilationException,
    EndOfArrayException,
    ParseException,
    ResolutionException,
)
from .jsontypes import JsonType
from .pointer import JsonPointer, RelativeJsonPointer


@dataclass
class Start:
    pass


@dataclass
class DoUp:
    n: int


@dataclass
class DoIndex:
    pass


@dataclass
class DoStep:
    step: str


Operation = Start | DoUp | DoIndex | DoStep


@dataclass
class JsonRef:
    doc: JsonType
    operation: Operation


@dataclass
class ResolveResult:
    refs: list[JsonRef]
    value: JsonType
    is_index_result: bool = False


def resolve(doc: JsonType, operations: list[Operation]) -> ResolveResult:
    '''

    >>> resolve({'foo': 1}, [Start(), DoStep('foo'), DoIndex(), DoIndex()])
    Traceback (most recent call last):
    fast_json_pointer.exceptions.ResolutionException: ...
    """
    '''
    cur_doc = doc
    refs: list[JsonRef] = [JsonRef(doc, Start())]

    if len(operations) == 0:
        return ResolveResult(refs, doc)

    last_op = operations[-1]

    for idx, op in enumerate(operations):
        match op:
            case DoUp(n):
                if n > 0:
                    refs = refs[:-n]
                    cur_doc = refs[-1].doc

            case DoStep(step):
                match cur_doc:
                    case dict():
                        if step not in cur_doc:
                            raise ResolutionException(
                                f"Key '{step}' not in JSON object",
                                refs=refs,
                                remaining=operations[idx:],
                            )

                        cur_doc = cur_doc[step]
                        refs.append(JsonRef(cur_doc, op))

                    case list():
                        if step == "-":
                            raise EndOfArrayException(
                                "Hit '-' (end of array) token",
                                refs=refs,
                                remaining=operations[idx:],
                            )

                        part_idx = int(step)
                        if part_idx >= len(cur_doc):
                            raise ResolutionException(
                                f"Index '{part_idx}' not in JSON array",
                                refs=refs,
                                remaining=operations[idx:],
                            )

                        cur_doc = cur_doc[part_idx]
                        refs.append(JsonRef(cur_doc, op))

                    case _:
                        raise ResolutionException(
                            f"Unnvaigable doc type '{type(step)}'",
                            refs=refs,
                            remaining=operations[idx:],
                        )

            case DoIndex():
                if op is not last_op:
                    raise ResolutionException(
                        f"Can't do 'index of' as anything other than the last operation",
                        refs=refs,
                        remaining=operations[idx:],
                    )

                return ResolveResult(
                    refs=refs, value=refs[-1].operation.step, is_index_result=True
                )

    return ResolveResult(refs, value=cur_doc)


@dataclass
class JsonResolver:
    operations: list[Operation]
    is_index_ref: bool

    @classmethod
    def compile(cls, *pointers: str | JsonPointer | RelativeJsonPointer) -> Self:
        ops = compile(*pointers)
        is_index_ref = isinstance(ops[-1], DoIndex) if ops else False

        return cls(ops, is_index_ref)

    def resolve(self, doc: JsonType) -> ResolveResult:
        return resolve(doc, self.operations)


def compile(*pointers: str | JsonPointer | RelativeJsonPointer) -> list[Operation]:
    """Builds a compilation plan for resolving a series of json pointers.

    Assumes regular json pointers should be resolved as a relative pointer w/ an offset
    of zero

    >>> compile('/foo', '0#', '0/bar')
    Traceback (most recent call last):
    fast_json_pointer.exceptions.CompilationException: ...
    """

    operations = []

    for idx, ptr in enumerate(pointers):
        match ptr:
            case str():
                try:
                    massaged = RelativeJsonPointer.parse(ptr)
                except ParseException:
                    massaged = JsonPointer.parse(ptr)

            case _:
                massaged = ptr

        match massaged:
            case JsonPointer(parts):
                operations.extend(DoStep(part) for part in parts)

            case RelativeJsonPointer(up, pointer):
                if up > 0:
                    operations.append(DoUp(up))

                if pointer is not None:
                    operations.extend(DoStep(part) for part in pointer.parts)

                if massaged.is_index_ref:
                    if (idx + 1) != len(pointers):
                        raise CompilationException(
                            "Can't follow an 'index of' json pointer with further "
                            "json pointers"
                        )

                    operations.append(DoIndex())

    return operations

1	from dataclasses import dataclass	1✔
2	from typing import *	1✔
3
4	from .exceptions import (	1✔
5	CompilationException,
6	EndOfArrayException,
7	ParseException,
8	ResolutionException,
9	)
10	from .jsontypes import JsonType	1✔
11	from .pointer import JsonPointer, RelativeJsonPointer	1✔
12
13
14	@dataclass	1✔
15	class Start:	1✔
16	pass	1✔
17
18
19	@dataclass	1✔
20	class DoUp:	1✔
21	n: int	1✔
22
23
24	@dataclass	1✔
25	class DoIndex:	1✔
26	pass	1✔
27
28
29	@dataclass	1✔
30	class DoStep:	1✔
31	step: str	1✔
32
33
34	Operation = Start \| DoUp \| DoIndex \| DoStep	1✔
35
36
37	@dataclass	1✔
38	class JsonRef:	1✔
39	doc: JsonType	1✔
40	operation: Operation	1✔
41
42
43	@dataclass	1✔
44	class ResolveResult:	1✔
45	refs: list[JsonRef]	1✔
46	value: JsonType	1✔
47	is_index_result: bool = False	1✔
48
49
50	def resolve(doc: JsonType, operations: list[Operation]) -> ResolveResult:	1✔
51	'''
52
53	>>> resolve({'foo': 1}, [Start(), DoStep('foo'), DoIndex(), DoIndex()])
54	Traceback (most recent call last):
55	fast_json_pointer.exceptions.ResolutionException: ...
56	"""
57	'''
58	cur_doc = doc	1✔
59	refs: list[JsonRef] = [JsonRef(doc, Start())]	1✔
60
61	if len(operations) == 0:	1✔
62	return ResolveResult(refs, doc)	1✔
63
64	last_op = operations[-1]	1✔
65
66	for idx, op in enumerate(operations):	1✔
67	match op:	1✔
68	case DoUp(n):	1✔
69	if n > 0:	1✔
70	refs = refs[:-n]	1✔
71	cur_doc = refs[-1].doc	1✔
72
73	case DoStep(step):	1✔
74	match cur_doc:	1✔
75	case dict():	1✔
76	if step not in cur_doc:	1✔
77	raise ResolutionException(	1✔
78	f"Key '{step}' not in JSON object",
79	refs=refs,
80	remaining=operations[idx:],
81	)
82
83	cur_doc = cur_doc[step]	1✔
84	refs.append(JsonRef(cur_doc, op))	1✔
85
86	case list():	1✔
87	if step == "-":	1✔
88	raise EndOfArrayException(	1✔
89	"Hit '-' (end of array) token",
90	refs=refs,
91	remaining=operations[idx:],
92	)
93
94	part_idx = int(step)	1✔
95	if part_idx >= len(cur_doc):	1✔
96	raise ResolutionException(	1✔
97	f"Index '{part_idx}' not in JSON array",
98	refs=refs,
99	remaining=operations[idx:],
100	)
101
102	cur_doc = cur_doc[part_idx]	1✔
103	refs.append(JsonRef(cur_doc, op))	1✔
104
105	case _:	1✔
106	raise ResolutionException(	1✔
107	f"Unnvaigable doc type '{type(step)}'",
108	refs=refs,
109	remaining=operations[idx:],
110	)
111
112	case DoIndex():	1✔
113	if op is not last_op:	1✔
114	raise ResolutionException(	1✔
115	f"Can't do 'index of' as anything other than the last operation",
116	refs=refs,
117	remaining=operations[idx:],
118	)
119
120	return ResolveResult(	1✔
121	refs=refs, value=refs[-1].operation.step, is_index_result=True
122	)
123
124	return ResolveResult(refs, value=cur_doc)	1✔
125
126
127	@dataclass	1✔
128	class JsonResolver:	1✔
129	operations: list[Operation]	1✔
130	is_index_ref: bool	1✔
131
132	@classmethod	1✔
133	def compile(cls, *pointers: str \| JsonPointer \| RelativeJsonPointer) -> Self:	1✔
134	ops = compile(*pointers)	1✔
135	is_index_ref = isinstance(ops[-1], DoIndex) if ops else False	1✔
136
137	return cls(ops, is_index_ref)	1✔
138
139	def resolve(self, doc: JsonType) -> ResolveResult:	1✔
140	return resolve(doc, self.operations)	1✔
141
142
143	def compile(*pointers: str \| JsonPointer \| RelativeJsonPointer) -> list[Operation]:	1✔
144	"""Builds a compilation plan for resolving a series of json pointers.
145
146	Assumes regular json pointers should be resolved as a relative pointer w/ an offset
147	of zero
148
149	>>> compile('/foo', '0#', '0/bar')
150	Traceback (most recent call last):
151	fast_json_pointer.exceptions.CompilationException: ...
152	"""
153
154	operations = []	1✔
155
156	for idx, ptr in enumerate(pointers):	1✔
157	match ptr:	1✔
158	case str():	1✔
159	try:	1✔
160	massaged = RelativeJsonPointer.parse(ptr)	1✔
161	except ParseException:	1✔
162	massaged = JsonPointer.parse(ptr)	1✔
163
164	case _:	1✔
165	massaged = ptr	1✔
166
167	match massaged:	1✔
168	case JsonPointer(parts):	1✔
169	operations.extend(DoStep(part) for part in parts)	1✔
170
171	case RelativeJsonPointer(up, pointer):	1✔
172	if up > 0:	1✔
173	operations.append(DoUp(up))	1✔
174
175	if pointer is not None:	1✔
176	operations.extend(DoStep(part) for part in pointer.parts)	1✔
177
178	if massaged.is_index_ref:	1✔
179	if (idx + 1) != len(pointers):	1✔
180	raise CompilationException(	1✔
181	"Can't follow an 'index of' json pointer with further "
182	"json pointers"
183	)
184
185	operations.append(DoIndex())	1✔
186
187	return operations	1✔

SlowAPI / fast-json-pointer / 3798344209

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