Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/boto3/resources/params.py: 15%

55 statements  

« prev     ^ index     » next       coverage.py v7.3.2, created at 2023-12-08 06:51 +0000

1# Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved. 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"). You 

4# may not use this file except in compliance with the License. A copy of 

5# the License is located at 

6# 

7# https://aws.amazon.com/apache2.0/ 

8# 

9# or in the "license" file accompanying this file. This file is 

10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF 

11# ANY KIND, either express or implied. See the License for the specific 

12# language governing permissions and limitations under the License. 

13 

14import re 

15 

16import jmespath 

17from botocore import xform_name 

18 

19from ..exceptions import ResourceLoadException 

20 

21INDEX_RE = re.compile(r'\[(.*)\]$') 

22 

23 

24def get_data_member(parent, path): 

25 """ 

26 Get a data member from a parent using a JMESPath search query, 

27 loading the parent if required. If the parent cannot be loaded 

28 and no data is present then an exception is raised. 

29 

30 :type parent: ServiceResource 

31 :param parent: The resource instance to which contains data we 

32 are interested in. 

33 :type path: string 

34 :param path: The JMESPath expression to query 

35 :raises ResourceLoadException: When no data is present and the 

36 resource cannot be loaded. 

37 :returns: The queried data or ``None``. 

38 """ 

39 # Ensure the parent has its data loaded, if possible. 

40 if parent.meta.data is None: 

41 if hasattr(parent, 'load'): 

42 parent.load() 

43 else: 

44 raise ResourceLoadException( 

45 f'{parent.__class__.__name__} has no load method!' 

46 ) 

47 

48 return jmespath.search(path, parent.meta.data) 

49 

50 

51def create_request_parameters(parent, request_model, params=None, index=None): 

52 """ 

53 Handle request parameters that can be filled in from identifiers, 

54 resource data members or constants. 

55 

56 By passing ``params``, you can invoke this method multiple times and 

57 build up a parameter dict over time, which is particularly useful 

58 for reverse JMESPath expressions that append to lists. 

59 

60 :type parent: ServiceResource 

61 :param parent: The resource instance to which this action is attached. 

62 :type request_model: :py:class:`~boto3.resources.model.Request` 

63 :param request_model: The action request model. 

64 :type params: dict 

65 :param params: If set, then add to this existing dict. It is both 

66 edited in-place and returned. 

67 :type index: int 

68 :param index: The position of an item within a list 

69 :rtype: dict 

70 :return: Pre-filled parameters to be sent to the request operation. 

71 """ 

72 if params is None: 

73 params = {} 

74 

75 for param in request_model.params: 

76 source = param.source 

77 target = param.target 

78 

79 if source == 'identifier': 

80 # Resource identifier, e.g. queue.url 

81 value = getattr(parent, xform_name(param.name)) 

82 elif source == 'data': 

83 # If this is a data member then it may incur a load 

84 # action before returning the value. 

85 value = get_data_member(parent, param.path) 

86 elif source in ['string', 'integer', 'boolean']: 

87 # These are hard-coded values in the definition 

88 value = param.value 

89 elif source == 'input': 

90 # This is provided by the user, so ignore it here 

91 continue 

92 else: 

93 raise NotImplementedError(f'Unsupported source type: {source}') 

94 

95 build_param_structure(params, target, value, index) 

96 

97 return params 

98 

99 

100def build_param_structure(params, target, value, index=None): 

101 """ 

102 This method provides a basic reverse JMESPath implementation that 

103 lets you go from a JMESPath-like string to a possibly deeply nested 

104 object. The ``params`` are mutated in-place, so subsequent calls 

105 can modify the same element by its index. 

106 

107 >>> build_param_structure(params, 'test[0]', 1) 

108 >>> print(params) 

109 {'test': [1]} 

110 

111 >>> build_param_structure(params, 'foo.bar[0].baz', 'hello world') 

112 >>> print(params) 

113 {'test': [1], 'foo': {'bar': [{'baz': 'hello, world'}]}} 

114 

115 """ 

116 pos = params 

117 parts = target.split('.') 

118 

119 # First, split into parts like 'foo', 'bar[0]', 'baz' and process 

120 # each piece. It can either be a list or a dict, depending on if 

121 # an index like `[0]` is present. We detect this via a regular 

122 # expression, and keep track of where we are in params via the 

123 # pos variable, walking down to the last item. Once there, we 

124 # set the value. 

125 for i, part in enumerate(parts): 

126 # Is it indexing an array? 

127 result = INDEX_RE.search(part) 

128 if result: 

129 if result.group(1): 

130 if result.group(1) == '*': 

131 part = part[:-3] 

132 else: 

133 # We have an explicit index 

134 index = int(result.group(1)) 

135 part = part[: -len(str(index) + '[]')] 

136 else: 

137 # Index will be set after we know the proper part 

138 # name and that it's a list instance. 

139 index = None 

140 part = part[:-2] 

141 

142 if part not in pos or not isinstance(pos[part], list): 

143 pos[part] = [] 

144 

145 # This means we should append, e.g. 'foo[]' 

146 if index is None: 

147 index = len(pos[part]) 

148 

149 while len(pos[part]) <= index: 

150 # Assume it's a dict until we set the final value below 

151 pos[part].append({}) 

152 

153 # Last item? Set the value, otherwise set the new position 

154 if i == len(parts) - 1: 

155 pos[part][index] = value 

156 else: 

157 # The new pos is the *item* in the array, not the array! 

158 pos = pos[part][index] 

159 else: 

160 if part not in pos: 

161 pos[part] = {} 

162 

163 # Last item? Set the value, otherwise set the new position 

164 if i == len(parts) - 1: 

165 pos[part] = value 

166 else: 

167 pos = pos[part]