def get_dns_name_and_fallback_mount_target_ip_address()

in src/mount_efs/__init__.py [0:0]

• 72 lines of code
• 14 McCabe index (conditional complexity)

def get_dns_name_and_fallback_mount_target_ip_address(config, fs_id, options):
    def _validate_replacement_field_count(format_str, expected_ct):
        if format_str.count("{") != expected_ct or format_str.count("}") != expected_ct:
            raise ValueError(
                "DNS name format has an incorrect number of replacement fields"
            )

    dns_name_format = config.get(CONFIG_SECTION, "dns_name_format")

    if "{fs_id}" not in dns_name_format:
        raise ValueError("DNS name format must include {fs_id}")

    format_args = {"fs_id": fs_id}

    expected_replacement_field_ct = 1

    if "{az}" in dns_name_format:
        az = options.get("az")
        if az:
            expected_replacement_field_ct += 1
            format_args["az"] = az
        else:
            dns_name_format = dns_name_format.replace("{az}.", "")

    if "{region}" in dns_name_format:
        expected_replacement_field_ct += 1
        format_args["region"] = get_target_region(config)

    if "{dns_name_suffix}" in dns_name_format:
        expected_replacement_field_ct += 1
        config_section = CONFIG_SECTION
        region = format_args.get("region")

        if region:
            config_section = get_config_section(config, region)

        format_args["dns_name_suffix"] = config.get(config_section, "dns_name_suffix")

        logging.debug(
            "Using dns_name_suffix %s in config section [%s]",
            format_args.get("dns_name_suffix"),
            config_section,
        )

    _validate_replacement_field_count(dns_name_format, expected_replacement_field_ct)

    dns_name = dns_name_format.format(**format_args)

    if "mounttargetip" in options:
        ip_address = options.get("mounttargetip")
        logging.info(
            "Use the mount target ip address %s provided in the mount options to mount."
            % ip_address
        )
        try:
            mount_target_ip_address_can_be_resolved(
                ip_address,
                passed_via_options=True,
                network_namespace=options.get("netns") if "netns" in options else None,
            )
            return dns_name, options.get("mounttargetip")
        except FallbackException as e:
            fallback_message = e.message
            throw_ip_address_connect_failure_with_fallback_message(
                ip_address=ip_address, fallback_message=fallback_message
            )

    if dns_name_can_be_resolved(dns_name):
        return dns_name, None

    logging.info(
        "Failed to resolve %s, attempting to lookup mount target ip address using botocore.",
        dns_name,
    )

    try:
        fallback_mount_target_ip_address = get_fallback_mount_target_ip_address(
            config, options, fs_id, dns_name
        )
        logging.info(
            "Found fall back mount target ip address %s for file system %s",
            fallback_mount_target_ip_address,
            fs_id,
        )
        return dns_name, fallback_mount_target_ip_address
    except FallbackException as e:
        fallback_message = e.message

    throw_dns_resolve_failure_with_fallback_message(dns_name, fallback_message)