def mark_statement_name_as_handled()

in lambda/python/rs_integration_function/ddb/ddb_state_table.py [0:0]

• 29 lines of code
• 1 McCabe index (conditional complexity)

    def mark_statement_name_as_handled(self, statement_name: StatementName, finished_event_details: dict) -> None:
        """
        We take the convention that if a TTL is set the statement_name has been processed. The TTL will allow automatic
        cleanup from DDB.
        Args:
            statement_name:
            finished_event_details: information reported by Data API finished event
        """
        ttl_field = self.get_ttl_value()

        logger.debug({
            l_statement_name: str(statement_name),
            l_finished_event_details: finished_event_details,
            l_ttl: ttl_field
        })
        response = self.update_item(
            Key={
                DDB_ID: statement_name.execution_arn,
                DDB_INVOCATION_ID: statement_name.invocation_id,
            },
            UpdateExpression="SET #T = :ttl, #D = :details",
            ReturnValues='ALL_OLD',
            ReturnConsumedCapacity='NONE',
            ExpressionAttributeNames={
                '#T': DDB_TTL,
                '#D': DDB_FINISHED_EVENT_DETAILS
            },
            ExpressionAttributeValues={
                ':ttl': ttl_field,
                ':details': finished_event_details
            }
        )
        logger.debug({
            l_statement_name: statement_name,
            l_response: response,
            l_ttl: ttl_field
        })