Coverage for src / check_datapackage / internals.py: 100%

1from dataclasses import dataclass 

2from itertools import chain 

3from typing import Annotated, Any, Callable, Iterable, TypeVar 

4 

5from jsonpath import ( 

6 CompoundJSONPath, 

7 JSONPathMatch, 

8 JSONPathSyntaxError, 

9 compile, 

10 finditer, 

11) 

12from pydantic import AfterValidator 

13 

14 

15@dataclass 

16class PropertyField: 

17 """The field of a Data Package property. 

18 

19 Attributes: 

20 jsonpath (str): The direct JSON path to the field. 

21 value (str): The value contained in the field. 

22 """ 

23 

24 jsonpath: str 

25 value: Any 

26 

27 

28def _get_fields_at_jsonpath( 

29 jsonpath: str, json_object: dict[str, Any] 

30) -> list[PropertyField]: 

31 """Returns all fields that match the JSON path.""" 

32 matches = finditer(jsonpath, json_object) 

33 return _map(matches, _create_property_field) 

34 

35 

36def _get_direct_jsonpaths(jsonpath: str, json_object: dict[str, Any]) -> list[str]: 

37 """Returns all direct JSON paths that match a direct or indirect JSON path.""" 

38 fields = _get_fields_at_jsonpath(jsonpath, json_object) 

39 return _map(fields, lambda field: field.jsonpath) 

40 

41 

42def _create_property_field(match: JSONPathMatch) -> PropertyField: 

43 return PropertyField( 

44 jsonpath=match.path.replace("['", ".").replace("']", ""), 

45 value=match.obj, 

46 ) 

47 

48 

49In = TypeVar("In") 

50Out = TypeVar("Out") 

51 

52 

53def _map(x: Iterable[In], fn: Callable[[In], Out]) -> list[Out]: 

54 return list(map(fn, x)) 

55 

56 

57def _filter(x: Iterable[In], fn: Callable[[In], bool]) -> list[In]: 

58 return list(filter(fn, x)) 

59 

60 

61def _flat_map(items: Iterable[In], fn: Callable[[In], Iterable[Out]]) -> list[Out]: 

62 """Maps and flattens the items by one level.""" 

63 return list(chain.from_iterable(map(fn, items))) 

64 

65 

66def _is_jsonpath(value: str) -> str: 

67 try: 

68 jsonpath = compile(value) 

69 except JSONPathSyntaxError: 

70 raise ValueError( 

71 f"'{value}' is not a correct JSON path. See " 

72 "https://jg-rp.github.io/python-jsonpath/syntax/ for the expected syntax." 

73 ) 

74 

75 # Doesn't allow intersection paths (e.g. `$.resources & $.name`). 

76 intersection_token = jsonpath.env.intersection_token 

77 if isinstance(jsonpath, CompoundJSONPath) and _filter( 

78 jsonpath.paths, lambda path: path[0] == intersection_token 

79 ): 

80 raise ValueError( 

81 f"The intersection operator (`{intersection_token}`) in the JSON path " 

82 f"'{value}' is not supported." 

83 ) 

84 return value 

85 

86 

87type JsonPath = Annotated[str, AfterValidator(_is_jsonpath)]