|
| 1 | +import datetime |
| 2 | +import logging |
| 3 | +from typing import Any, Dict, Optional |
| 4 | + |
| 5 | +import boto3 |
| 6 | +from botocore.config import Config |
| 7 | + |
| 8 | +from aws_lambda_powertools.utilities.idempotency import BasePersistenceLayer |
| 9 | +from aws_lambda_powertools.utilities.idempotency.exceptions import ( |
| 10 | + IdempotencyItemAlreadyExistsError, |
| 11 | + IdempotencyItemNotFoundError, |
| 12 | +) |
| 13 | +from aws_lambda_powertools.utilities.idempotency.persistence.base import DataRecord |
| 14 | + |
| 15 | +logger = logging.getLogger(__name__) |
| 16 | + |
| 17 | + |
| 18 | +class DynamoDBPersistenceLayer(BasePersistenceLayer): |
| 19 | + def __init__( |
| 20 | + self, |
| 21 | + table_name: str, |
| 22 | + key_attr: str = "id", |
| 23 | + expiry_attr: str = "expiration", |
| 24 | + status_attr: str = "status", |
| 25 | + data_attr: str = "data", |
| 26 | + validation_key_attr: str = "validation", |
| 27 | + boto_config: Optional[Config] = None, |
| 28 | + boto3_session: Optional[boto3.session.Session] = None, |
| 29 | + ): |
| 30 | + boto_config = boto_config or Config() |
| 31 | + session = boto3_session or boto3.session.Session() |
| 32 | + self._ddb_resource = session.resource("dynamodb", config=boto_config) |
| 33 | + self.table_name = table_name |
| 34 | + self.table = self._ddb_resource.Table(self.table_name) |
| 35 | + self.key_attr = key_attr |
| 36 | + self.expiry_attr = expiry_attr |
| 37 | + self.status_attr = status_attr |
| 38 | + self.data_attr = data_attr |
| 39 | + self.validation_key_attr = validation_key_attr |
| 40 | + super(DynamoDBPersistenceLayer, self).__init__() |
| 41 | + |
| 42 | + def _item_to_data_record(self, item: Dict[str, Any]) -> DataRecord: |
| 43 | + """ |
| 44 | + Translate raw item records from DynamoDB to DataRecord |
| 45 | +
|
| 46 | + Parameters |
| 47 | + ---------- |
| 48 | + item: Dict[str, Union[str, int]] |
| 49 | + Item format from dynamodb response |
| 50 | +
|
| 51 | + Returns |
| 52 | + ------- |
| 53 | + DataRecord |
| 54 | + representation of item |
| 55 | +
|
| 56 | + """ |
| 57 | + return DataRecord( |
| 58 | + idempotency_key=item[self.key_attr], |
| 59 | + status=item[self.status_attr], |
| 60 | + expiry_timestamp=item[self.expiry_attr], |
| 61 | + response_data=item.get(self.data_attr), |
| 62 | + payload_hash=item.get(self.validation_key_attr), |
| 63 | + ) |
| 64 | + |
| 65 | + def _get_record(self, idempotency_key) -> DataRecord: |
| 66 | + response = self.table.get_item(Key={self.key_attr: idempotency_key}, ConsistentRead=True) |
| 67 | + |
| 68 | + try: |
| 69 | + item = response["Item"] |
| 70 | + except KeyError: |
| 71 | + raise IdempotencyItemNotFoundError |
| 72 | + return self._item_to_data_record(item) |
| 73 | + |
| 74 | + def _put_record(self, data_record: DataRecord) -> None: |
| 75 | + item = { |
| 76 | + self.key_attr: data_record.idempotency_key, |
| 77 | + self.expiry_attr: data_record.expiry_timestamp, |
| 78 | + self.status_attr: data_record.status, |
| 79 | + } |
| 80 | + |
| 81 | + if self.payload_validation_enabled: |
| 82 | + item[self.validation_key_attr] = data_record.payload_hash |
| 83 | + |
| 84 | + now = datetime.datetime.now() |
| 85 | + try: |
| 86 | + logger.debug(f"Putting record for idempotency key: {data_record.idempotency_key}") |
| 87 | + self.table.put_item( |
| 88 | + Item=item, |
| 89 | + ConditionExpression=f"attribute_not_exists({self.key_attr}) OR {self.expiry_attr} < :now", |
| 90 | + ExpressionAttributeValues={":now": int(now.timestamp())}, |
| 91 | + ) |
| 92 | + except self._ddb_resource.meta.client.exceptions.ConditionalCheckFailedException: |
| 93 | + logger.debug(f"Failed to put record for already existing idempotency key: {data_record.idempotency_key}") |
| 94 | + raise IdempotencyItemAlreadyExistsError |
| 95 | + |
| 96 | + def _update_record(self, data_record: DataRecord): |
| 97 | + logger.debug(f"Updating record for idempotency key: {data_record.idempotency_key}") |
| 98 | + update_expression = "SET #response_data = :response_data, #expiry = :expiry, #status = :status" |
| 99 | + expression_attr_values = { |
| 100 | + ":expiry": data_record.expiry_timestamp, |
| 101 | + ":response_data": data_record.response_data, |
| 102 | + ":status": data_record.status, |
| 103 | + } |
| 104 | + expression_attr_names = { |
| 105 | + "#response_data": self.data_attr, |
| 106 | + "#expiry": self.expiry_attr, |
| 107 | + "#status": self.status_attr, |
| 108 | + } |
| 109 | + |
| 110 | + if self.payload_validation_enabled: |
| 111 | + update_expression += ", #validation_key = :validation_key" |
| 112 | + expression_attr_values[":validation_key"] = data_record.payload_hash |
| 113 | + expression_attr_names["#validation_key"] = self.validation_key_attr |
| 114 | + |
| 115 | + kwargs = { |
| 116 | + "Key": {self.key_attr: data_record.idempotency_key}, |
| 117 | + "UpdateExpression": update_expression, |
| 118 | + "ExpressionAttributeValues": expression_attr_values, |
| 119 | + "ExpressionAttributeNames": expression_attr_names, |
| 120 | + } |
| 121 | + |
| 122 | + self.table.update_item(**kwargs) |
| 123 | + |
| 124 | + def _delete_record(self, data_record: DataRecord) -> None: |
| 125 | + logger.debug(f"Deleting record for idempotency key: {data_record.idempotency_key}") |
| 126 | + self.table.delete_item(Key={self.key_attr: data_record.idempotency_key}) |
0 commit comments