def parse_get_bucket_replication_progress_result()

in oss2/xml_utils.py [0:0]

• 24 lines of code
• 4 McCabe index (conditional complexity)

def parse_get_bucket_replication_progress_result(result, body):
    root = ElementTree.fromstring(body)

    rule_node = root.find("Rule")
    progress = BucketReplicationProgress()
    progress.rule_id = _find_tag(rule_node, "ID")

    destination_node = rule_node.find("Destination")
    progress.target_bucket_name = _find_tag(destination_node, "Bucket")
    progress.target_bucket_location = _find_tag(destination_node, "Location")
    progress.target_transfer_type = _find_tag_with_default(destination_node, "TransferType", None)

    progress.status = _find_tag(rule_node, "Status")

    if _find_tag(rule_node, "HistoricalObjectReplication") == 'enabled':
        progress.is_enable_historical_object_replication = True
    else:
        progress.is_enable_historical_object_replication = False

    prefixes_node = rule_node.find('PrefixSet')
    if prefixes_node is not None:
        progress.prefix_list = _find_all_tags(prefixes_node, 'Prefix')

    actions = _find_tag(rule_node, 'Action')
    progress.action_list = actions.split(',')

    historical_object_progress = _find_tag_with_default(rule_node, 'Progress/HistoricalObject', None)
    if historical_object_progress is not None:
        progress.historical_object_progress = float(historical_object_progress)
    progress.new_object_progress = _find_tag_with_default(rule_node, 'Progress/NewObject', None)

    result.progress = progress