Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/fastjsonschema/ref_resolver.py: 23%

1# pylint: disable=import-outside-toplevel 

2 

3""" 

4JSON Schema URI resolution scopes and dereferencing 

5 

6https://tools.ietf.org/id/draft-zyp-json-schema-04.html#rfc.section.7 

7 

8Code adapted from https://github.com/Julian/jsonschema 

9""" 

10 

11import contextlib 

12import json 

13import re 

14import sys 

15from collections.abc import Callable, Iterator, Mapping 

16from typing import Any 

17from urllib import parse as urlparse 

18from urllib.parse import unquote 

19 

20from .exceptions import JsonSchemaDefinitionException 

21 

22MAX_SCHEMA_WALK_DEPTH = min(500, sys.getrecursionlimit() // 2) 

23 

24 

25def get_id(schema: dict[str, Any]) -> str: 

26 """ 

27 Originally ID was `id` and since v7 it's `$id`. 

28 """ 

29 return schema.get('$id', schema.get('id', '')) 

30 

31 

32def resolve_path(schema: Any, fragment: str) -> Any: 

33 """ 

34 Return definition from path. 

35 

36 Path is unescaped according https://tools.ietf.org/html/rfc6901 

37 """ 

38 fragment = fragment.lstrip('/') 

39 parts = unquote(fragment).split('/') if fragment else [] 

40 for part in parts: 

41 part = part.replace('~1', '/').replace('~0', '~') 

42 if isinstance(schema, list): 

43 schema = schema[int(part)] 

44 elif part in schema: 

45 schema = schema[part] 

46 else: 

47 raise JsonSchemaDefinitionException('Unresolvable ref: {}'.format(part)) 

48 return schema 

49 

50 

51def normalize(uri: str) -> str: 

52 return urlparse.urlsplit(uri).geturl() 

53 

54 

55def resolve_remote(uri: str, handlers: Mapping[str, Callable[[str], Any]]) -> Any: 

56 """ 

57 Resolve a remote ``uri``. 

58 

59 .. note:: 

60 

61 urllib library is used to fetch requests from the remote ``uri`` 

62 if handlers does notdefine otherwise. 

63 """ 

64 scheme = urlparse.urlsplit(uri).scheme 

65 if scheme in handlers: 

66 result = handlers[scheme](uri) 

67 else: 

68 from urllib.request import urlopen 

69 

70 with urlopen(uri) as response: 

71 encoding = response.info().get_content_charset() or 'utf-8' 

72 try: 

73 result = json.loads(response.read().decode(encoding),) 

74 except ValueError as exc: 

75 raise JsonSchemaDefinitionException('{} failed to decode'.format(uri)) from exc 

76 return result 

77 

78 

79class RefResolver: 

80 """ 

81 Resolve JSON References. 

82 """ 

83 

84 # pylint: disable=dangerous-default-value,too-many-arguments 

85 def __init__( 

86 self, 

87 base_uri: str, 

88 schema: dict[str, Any] | bool, 

89 store: dict[str, Any] = {}, 

90 cache: bool = True, 

91 handlers: Mapping[str, Callable[[str], Any]] = {}, 

92 ) -> None: 

93 """ 

94 `base_uri` is URI of the referring document from the `schema`. 

95 `store` is an dictionary that will be used to cache the fetched schemas 

96 (if `cache=True`). 

97 

98 Please notice that you can have caching problems when compiling schemas 

99 with colliding `$ref`. To force overwriting use `cache=False` or 

100 explicitly pass the `store` argument (with a brand new dictionary) 

101 """ 

102 self.base_uri = base_uri 

103 self.resolution_scope = base_uri 

104 self.schema = schema 

105 self.store = store 

106 self.cache = cache 

107 self.handlers = handlers 

108 self._walked_uris: set[str] = set() 

109 self.walk(schema) 

110 self._walked_uris.add(normalize(base_uri) if base_uri else '') 

111 

112 @classmethod 

113 def from_schema( 

114 cls, 

115 schema: dict[str, Any] | bool, 

116 handlers: Mapping[str, Callable[[str], Any]] = {}, 

117 **kwargs: Any, 

118 ) -> 'RefResolver': 

119 """ 

120 Construct a resolver from a JSON schema object. 

121 """ 

122 return cls( 

123 get_id(schema) if isinstance(schema, dict) else '', 

124 schema, 

125 handlers=handlers, 

126 **kwargs 

127 ) 

128 

129 @contextlib.contextmanager 

130 def in_scope(self, scope: str) -> Iterator[None]: 

131 """ 

132 Context manager to handle current scope. 

133 """ 

134 old_scope = self.resolution_scope 

135 self.resolution_scope = urlparse.urljoin(old_scope, scope) 

136 try: 

137 yield 

138 finally: 

139 self.resolution_scope = old_scope 

140 

141 @contextlib.contextmanager 

142 def resolving(self, ref: str) -> Iterator[Any]: 

143 """ 

144 Context manager which resolves a JSON ``ref`` and enters the 

145 resolution scope of this ref. 

146 """ 

147 new_uri = urlparse.urljoin(self.resolution_scope, ref) 

148 uri, fragment = urlparse.urldefrag(new_uri) 

149 

150 document_uri = uri or self.base_uri 

151 

152 if uri and normalize(uri) in self.store: 

153 schema = self.store[normalize(uri)] 

154 elif not uri or uri == self.base_uri: 

155 schema = self.schema 

156 else: 

157 schema = resolve_remote(uri, self.handlers) 

158 if self.cache: 

159 self.store[normalize(uri)] = schema 

160 

161 old_base_uri, old_schema = self.base_uri, self.schema 

162 self.base_uri, self.schema = document_uri, schema 

163 try: 

164 with self.in_scope(document_uri): 

165 self._ensure_walked(document_uri, schema) 

166 if fragment and not fragment.startswith('/'): 

167 plain_name = normalize(urlparse.urljoin(document_uri, '#' + fragment)) 

168 if plain_name in self.store: 

169 yield self.store[plain_name] 

170 return 

171 raise JsonSchemaDefinitionException('Unresolvable ref: {}'.format(fragment)) 

172 yield resolve_path(schema, fragment) 

173 finally: 

174 self.base_uri, self.schema = old_base_uri, old_schema 

175 

176 def _ensure_walked(self, uri: str, schema: Any) -> None: 

177 normalized = normalize(uri) if uri else '' 

178 if normalized in self._walked_uris: 

179 return 

180 self.walk(schema, rewrite_refs=False) 

181 self._walked_uris.add(normalized) 

182 

183 def get_uri(self) -> str: 

184 return normalize(self.resolution_scope) 

185 

186 def get_scope_name(self) -> str: 

187 """ 

188 Get current scope and return it as a valid function name. 

189 """ 

190 name = 'validate_' + unquote(self.resolution_scope).replace('~1', '_').replace('~0', '_').replace('"', '') 

191 name = re.sub(r'($[^a-zA-Z]|[^a-zA-Z0-9])', '_', name) 

192 name = name.lower().rstrip('_') 

193 return name 

194 

195 def walk(self, node: dict[str, Any] | bool, depth: int = 0, rewrite_refs: bool = True) -> None: 

196 """ 

197 Walk thru schema and dereferencing ``id`` and ``$ref`` instances 

198 """ 

199 if depth >= MAX_SCHEMA_WALK_DEPTH: 

200 raise JsonSchemaDefinitionException( 

201 'Schema is too deeply nested (maximum depth is {})'.format(MAX_SCHEMA_WALK_DEPTH) 

202 ) 

203 

204 if isinstance(node, bool): 

205 pass 

206 elif '$ref' in node and isinstance(node['$ref'], str): 

207 if rewrite_refs: 

208 ref = node['$ref'] 

209 node['$ref'] = urlparse.urljoin(self.resolution_scope, ref) 

210 elif ('$id' in node or 'id' in node) and isinstance(get_id(node), str): 

211 with self.in_scope(get_id(node)): 

212 self.store[normalize(self.resolution_scope)] = node 

213 for _, item in node.items(): 

214 if isinstance(item, dict): 

215 self.walk(item, depth + 1, rewrite_refs) 

216 else: 

217 for _, item in node.items(): 

218 if isinstance(item, dict): 

219 self.walk(item, depth + 1, rewrite_refs)