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

1from dataclasses import dataclass 

2from typing import Annotated, Any 

3 

4from jsonpath import ( 

5 CompoundJSONPath, 

6 JSONPathMatch, 

7 JSONPathSyntaxError, 

8 compile, 

9 finditer, 

10) 

11from pydantic import AfterValidator 

12from seedcase_soil import fmap, keep 

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 fmap(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 fmap(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 

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

50 try: 

51 jsonpath = compile(value) 

52 except JSONPathSyntaxError: 

53 raise ValueError( 

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

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

56 ) 

57 

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

59 intersection_token = jsonpath.env.intersection_token 

60 if isinstance(jsonpath, CompoundJSONPath) and keep( 

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

62 ): 

63 raise ValueError( 

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

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

66 ) 

67 return value 

68 

69 

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