def register_execution_start()

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

• 16 lines of code
• 2 McCabe index (conditional complexity)

    def register_execution_start(self, callback_object: CallbackInterface, sql_statement: str) -> StatementName:
        """
        Register a UUID4 string in a state table in DynamoDB and link it with the task of the stepfunction execution.
        Return this GUID string such that it can be used as statement name to update the task when the statement
        completes.
        """
        statement_name = StatementName.from_execution_arn(callback_object.get_id())
        item_details = {
            DDB_ID: statement_name.execution_arn,
            DDB_INVOCATION_ID: statement_name.invocation_id,
            SQL_STATEMENT: sql_statement,
            DDB_CALLBACK_DETAILS: callback_object.to_json()
        }
        if isinstance(callback_object, NoCallback):
            # No callback is expected so TTL can immediately be set.
            item_details[DDB_TTL] = self.get_ttl_value()
        logger.debug({l_item: item_details})
        self.put_item(
            Item=item_details,
            ConditionExpression="attribute_not_exists(sqlStatement)"  # Re-registration is  not allowed
        )
        return statement_name