Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/coreapi/client.py: 14%

1from coreapi import codecs, exceptions, transports 

2from coreapi.compat import string_types 

3from coreapi.document import Document, Link 

4from coreapi.utils import determine_transport, get_installed_codecs 

5import collections 

6import itypes 

7 

8 

9LinkAncestor = collections.namedtuple('LinkAncestor', ['document', 'keys']) 

10 

11 

12def _lookup_link(document, keys): 

13 """ 

14 Validates that keys looking up a link are correct. 

15 

16 Returns a two-tuple of (link, link_ancestors). 

17 """ 

18 if not isinstance(keys, (list, tuple)): 

19 msg = "'keys' must be a list of strings or ints." 

20 raise TypeError(msg) 

21 if any([ 

22 not isinstance(key, string_types) and not isinstance(key, int) 

23 for key in keys 

24 ]): 

25 raise TypeError("'keys' must be a list of strings or ints.") 

26 

27 # Determine the link node being acted on, and its parent document. 

28 # 'node' is the link we're calling the action for. 

29 # 'document_keys' is the list of keys to the link's parent document. 

30 node = document 

31 link_ancestors = [LinkAncestor(document=document, keys=[])] 

32 for idx, key in enumerate(keys): 

33 try: 

34 node = node[key] 

35 except (KeyError, IndexError, TypeError): 

36 index_string = ''.join('[%s]' % repr(key).strip('u') for key in keys) 

37 msg = 'Index %s did not reference a link. Key %s was not found.' 

38 raise exceptions.LinkLookupError(msg % (index_string, repr(key).strip('u'))) 

39 if isinstance(node, Document): 

40 ancestor = LinkAncestor(document=node, keys=keys[:idx + 1]) 

41 link_ancestors.append(ancestor) 

42 

43 # Ensure that we've correctly indexed into a link. 

44 if not isinstance(node, Link): 

45 index_string = ''.join('[%s]' % repr(key).strip('u') for key in keys) 

46 msg = "Can only call 'action' on a Link. Index %s returned type '%s'." 

47 raise exceptions.LinkLookupError( 

48 msg % (index_string, type(node).__name__) 

49 ) 

50 

51 return (node, link_ancestors) 

52 

53 

54def _validate_parameters(link, parameters): 

55 """ 

56 Ensure that parameters passed to the link are correct. 

57 Raises a `ParameterError` if any parameters do not validate. 

58 """ 

59 provided = set(parameters.keys()) 

60 required = set([ 

61 field.name for field in link.fields if field.required 

62 ]) 

63 optional = set([ 

64 field.name for field in link.fields if not field.required 

65 ]) 

66 

67 errors = {} 

68 

69 # Determine if any required field names not supplied. 

70 missing = required - provided 

71 for item in missing: 

72 errors[item] = 'This parameter is required.' 

73 

74 # Determine any parameter names supplied that are not valid. 

75 unexpected = provided - (optional | required) 

76 for item in unexpected: 

77 errors[item] = 'Unknown parameter.' 

78 

79 if errors: 

80 raise exceptions.ParameterError(errors) 

81 

82 

83def get_default_decoders(): 

84 return [ 

85 codecs.CoreJSONCodec(), 

86 codecs.JSONCodec(), 

87 codecs.TextCodec(), 

88 codecs.DownloadCodec() 

89 ] 

90 

91 

92def get_default_transports(auth=None, session=None): 

93 return [ 

94 transports.HTTPTransport(auth=auth, session=session) 

95 ] 

96 

97 

98class Client(itypes.Object): 

99 def __init__(self, decoders=None, transports=None, auth=None, session=None): 

100 assert transports is None or auth is None, ( 

101 "Cannot specify both 'auth' and 'transports'. " 

102 "When specifying transport instances explicitly you should set " 

103 "the authentication directly on the transport." 

104 ) 

105 if decoders is None: 

106 decoders = get_default_decoders() 

107 if transports is None: 

108 transports = get_default_transports(auth=auth) 

109 self._decoders = itypes.List(decoders) 

110 self._transports = itypes.List(transports) 

111 

112 @property 

113 def decoders(self): 

114 return self._decoders 

115 

116 @property 

117 def transports(self): 

118 return self._transports 

119 

120 def get(self, url, format=None, force_codec=False): 

121 link = Link(url, action='get') 

122 

123 decoders = self.decoders 

124 if format: 

125 force_codec = True 

126 decoders = [decoder for decoder in self.decoders if decoder.format == format] 

127 if not decoders: 

128 installed_codecs = get_installed_codecs() 

129 if format in installed_codecs: 

130 decoders = [installed_codecs[format]] 

131 else: 

132 raise ValueError("No decoder available with format='%s'" % format) 

133 

134 # Perform the action, and return a new document. 

135 transport = determine_transport(self.transports, link.url) 

136 return transport.transition(link, decoders, force_codec=force_codec) 

137 

138 def reload(self, document, format=None, force_codec=False): 

139 # Fallback for v1.x. To be removed in favour of explict `get` style. 

140 return self.get(document.url, format=format, force_codec=force_codec) 

141 

142 def action(self, document, keys, params=None, validate=True, overrides=None, 

143 action=None, encoding=None, transform=None): 

144 if (action is not None) or (encoding is not None) or (transform is not None): 

145 # Fallback for v1.x overrides. 

146 # Will be removed at some point, most likely in a 2.1 release. 

147 if overrides is None: 

148 overrides = {} 

149 if action is not None: 

150 overrides['action'] = action 

151 if encoding is not None: 

152 overrides['encoding'] = encoding 

153 if transform is not None: 

154 overrides['transform'] = transform 

155 

156 if isinstance(keys, string_types): 

157 keys = [keys] 

158 

159 if params is None: 

160 params = {} 

161 

162 # Validate the keys and link parameters. 

163 link, link_ancestors = _lookup_link(document, keys) 

164 if validate: 

165 _validate_parameters(link, params) 

166 

167 if overrides: 

168 # Handle any explicit overrides. 

169 url = overrides.get('url', link.url) 

170 action = overrides.get('action', link.action) 

171 encoding = overrides.get('encoding', link.encoding) 

172 transform = overrides.get('transform', link.transform) 

173 fields = overrides.get('fields', link.fields) 

174 link = Link(url, action=action, encoding=encoding, transform=transform, fields=fields) 

175 

176 # Perform the action, and return a new document. 

177 transport = determine_transport(self.transports, link.url) 

178 return transport.transition(link, self.decoders, params=params, link_ancestors=link_ancestors)