Coverage for dynamodx / repository.py: 27%

1from typing import TYPE_CHECKING, Any, cast 

2 

3from .transact_writer import TransactWriter 

4from .types import deserialize, serialize 

5 

6if TYPE_CHECKING: 

7 from mypy_boto3_dynamodb.client import DynamoDBClient 

8 from mypy_boto3_dynamodb.literals import ( 

9 ReturnValuesOnConditionCheckFailureType, 

10 ReturnValueType, 

11 SelectType, 

12 ) 

13 from mypy_boto3_dynamodb.type_defs import ( 

14 DeleteItemOutputTypeDef, 

15 GetItemOutputTypeDef, 

16 PutItemOutputTypeDef, 

17 QueryOutputTypeDef, 

18 UpdateItemOutputTypeDef, 

19 ) 

20else: 

21 DynamoDBClient = Any 

22 ReturnValueType = Any 

23 ReturnValuesOnConditionCheckFailureType = Any 

24 DeleteItemOutputTypeDef = Any 

25 QueryOutputTypeDef = Any 

26 GetItemOutputTypeDef = Any 

27 PutItemOutputTypeDef = Any 

28 UpdateItemOutputTypeDef = Any 

29 

30 

31class Repository: 

32 def __init__( 

33 self, 

34 table_name: str, 

35 *, 

36 client: DynamoDBClient, 

37 ) -> None: 

38 self._table_name = table_name 

39 self._client = client 

40 

41 def query( 

42 self, 

43 key_cond_expr: str, 

44 *, 

45 select: SelectType | None = None, 

46 expr_attr_names: dict | None = None, 

47 expr_attr_values: dict | None = None, 

48 filter_expr: str | None = None, 

49 projection_expr: str | None = None, 

50 limit: int | None = None, 

51 scan_index_forward: bool = True, 

52 exclusive_start_key: dict | None = None, 

53 table_name: str | None = None, 

54 ) -> QueryOutputTypeDef: 

55 """You must provide the name of the partition key attribute 

56 and a single value for that attribute. 

57 

58 Query returns all items with that partition key value. 

59 Optionally, you can provide a sort key attribute and use a comparison operator 

60 to refine the search results. 

61 

62 ... 

63 

64 A `Query` operation always returns a result set. If no matching items are found, 

65 the result set will be empty. 

66 Queries that do not return results consume the minimum number 

67 of read capacity units for that type of read operation. 

68 

69 - https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb/client/query.html 

70 """ 

71 attrs: dict = { 

72 'TableName': table_name or self._table_name, 

73 'KeyConditionExpression': key_cond_expr, 

74 'ScanIndexForward': scan_index_forward, 

75 } 

76 

77 if select: 

78 attrs['Select'] = select 

79 

80 if limit: 

81 attrs['Limit'] = limit 

82 

83 if expr_attr_names: 

84 attrs['ExpressionAttributeNames'] = expr_attr_names 

85 

86 if expr_attr_values: 

87 attrs['ExpressionAttributeValues'] = serialize(expr_attr_values) 

88 

89 if exclusive_start_key: 

90 attrs['ExclusiveStartKey'] = serialize(exclusive_start_key) 

91 

92 if filter_expr: 

93 attrs['FilterExpression'] = filter_expr 

94 

95 if projection_expr: 

96 attrs['ProjectionExpression'] = projection_expr 

97 

98 output = self._client.query(**attrs) 

99 

100 return cast( 

101 QueryOutputTypeDef, 

102 output 

103 | dict( 

104 Items=[deserialize(item) for item in output.get('Items', [])], 

105 ), 

106 ) 

107 

108 def get_item( 

109 self, 

110 key: dict, 

111 *, 

112 table_name: str | None = None, 

113 expr_attr_names: dict | None = None, 

114 projection_expr: str | None = None, 

115 ) -> GetItemOutputTypeDef: 

116 """The GetItem operation returns a set of attributes for the item 

117 with the given primary key. 

118 

119 If there is no matching item, GetItem does not return any data and 

120 there will be no Item element in the response. 

121 

122 - https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb/client/get_item.html 

123 """ 

124 attrs = { 

125 'TableName': table_name or self._table_name, 

126 'Key': serialize(key), 

127 } 

128 

129 if expr_attr_names: 

130 attrs['ExpressionAttributeNames'] = expr_attr_names 

131 

132 if projection_expr: 

133 attrs['ProjectionExpression'] = projection_expr 

134 

135 output = self._client.get_item(**attrs) 

136 

137 return cast( 

138 GetItemOutputTypeDef, 

139 output 

140 | { 

141 'Item': deserialize(output.get('Item', {})), 

142 }, 

143 ) 

144 

145 def put_item( 

146 self, 

147 item: dict, 

148 *, 

149 cond_expr: str | None = None, 

150 expr_attr_names: dict | None = None, 

151 expr_attr_values: dict | None = None, 

152 table_name: str | None = None, 

153 return_values: ReturnValueType | None = None, 

154 return_on_cond_fail: ReturnValuesOnConditionCheckFailureType | None = None, 

155 ) -> PutItemOutputTypeDef: 

156 attrs = { 

157 'TableName': table_name or self._table_name, 

158 'Item': serialize(item), 

159 } 

160 

161 if cond_expr: 

162 attrs['ConditionExpression'] = cond_expr 

163 

164 if expr_attr_names: 

165 attrs['ExpressionAttributeNames'] = expr_attr_names 

166 

167 if expr_attr_values: 

168 attrs['ExpressionAttributeValues'] = serialize(expr_attr_values) 

169 

170 if return_values: 

171 attrs['ReturnValues'] = return_values 

172 

173 if return_on_cond_fail: 

174 attrs['ReturnValuesOnConditionCheckFailure'] = return_on_cond_fail 

175 

176 return self._client.put_item(**attrs) 

177 

178 def update_item( 

179 self, 

180 key: dict, 

181 *, 

182 update_expr: str, 

183 cond_expr: str | None = None, 

184 expr_attr_names: dict | None = None, 

185 expr_attr_values: dict | None = None, 

186 table_name: str | None = None, 

187 return_values: ReturnValueType | None = None, 

188 return_on_cond_fail: ReturnValuesOnConditionCheckFailureType | None = None, 

189 ) -> UpdateItemOutputTypeDef: 

190 attrs: dict = { 

191 'TableName': table_name or self._table_name, 

192 'Key': serialize(key), 

193 'UpdateExpression': update_expr, 

194 } 

195 

196 if cond_expr: 

197 attrs['ConditionExpression'] = cond_expr 

198 

199 if expr_attr_names: 

200 attrs['ExpressionAttributeNames'] = expr_attr_names 

201 

202 if expr_attr_values: 

203 attrs['ExpressionAttributeValues'] = serialize(expr_attr_values) 

204 

205 if return_values: 

206 attrs['ReturnValues'] = return_values 

207 

208 if return_on_cond_fail: 

209 attrs['ReturnValuesOnConditionCheckFailure'] = return_on_cond_fail 

210 

211 return self._client.update_item(**attrs) 

212 

213 def delete_item( 

214 self, 

215 key: dict, 

216 *, 

217 cond_expr: str | None = None, 

218 expr_attr_names: dict | None = None, 

219 expr_attr_values: dict | None = None, 

220 table_name: str | None = None, 

221 return_on_cond_fail: ReturnValuesOnConditionCheckFailureType | None = None, 

222 ) -> DeleteItemOutputTypeDef: 

223 """Deletes a single item in a table by primary key. You can perform 

224 a conditional delete operation that deletes the item if it exists, 

225 or if it has an expected attribute value. 

226 """ 

227 attrs: dict = { 

228 'TableName': table_name or self._table_name, 

229 'Key': serialize(key), 

230 } 

231 

232 if cond_expr: 

233 attrs['ConditionExpression'] = cond_expr 

234 

235 if expr_attr_names: 

236 attrs['ExpressionAttributeNames'] = expr_attr_names 

237 

238 if expr_attr_values: 

239 attrs['ExpressionAttributeValues'] = serialize(expr_attr_values) 

240 

241 if return_on_cond_fail: 

242 attrs['ReturnValuesOnConditionCheckFailure'] = return_on_cond_fail 

243 

244 return self._client.delete_item(**attrs) 

245 

246 def transact_writer( 

247 self, 

248 flush_amount: int = 50, 

249 table_name: str | None = None, 

250 ) -> TransactWriter: 

251 return TransactWriter( 

252 table_name=table_name or self._table_name, 

253 client=self._client, 

254 flush_amount=flush_amount, 

255 ) 

256 

257 

258DynamoDBRepository = Repository