def _put_record()

in aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py [0:0]

• 20 lines of code
• 3 McCabe index (conditional complexity)

    def _put_record(self, data_record: DataRecord) -> None:
        item = {
            **self._get_key(data_record.idempotency_key),
            self.expiry_attr: data_record.expiry_timestamp,
            self.status_attr: data_record.status,
        }

        if self.payload_validation_enabled:
            item[self.validation_key_attr] = data_record.payload_hash

        now = datetime.datetime.now()
        try:
            logger.debug(f"Putting record for idempotency key: {data_record.idempotency_key}")
            self.table.put_item(
                Item=item,
                ConditionExpression="attribute_not_exists(#id) OR #now < :now",
                ExpressionAttributeNames={"#id": self.key_attr, "#now": self.expiry_attr},
                ExpressionAttributeValues={":now": int(now.timestamp())},
            )
        except self.table.meta.client.exceptions.ConditionalCheckFailedException:
            logger.debug(f"Failed to put record for already existing idempotency key: {data_record.idempotency_key}")
            raise IdempotencyItemAlreadyExistsError