def change_txt_record()

in lemur/plugins/lemur_acme/route53.py [0:0]

• 46 lines of code
• 11 McCabe index (conditional complexity)

def change_txt_record(action, zone_id, domain, value, client=None):
    current_txt_records = []
    try:
        current_records = client.list_resource_record_sets(
            HostedZoneId=zone_id,
            StartRecordName=domain,
            StartRecordType="TXT",
            MaxItems="1",
        )["ResourceRecordSets"]

        for record in current_records:
            if record.get("Type") == "TXT":
                current_txt_records.extend(record.get("ResourceRecords", []))
    except Exception as e:
        # Current Resource Record does not exist
        if "NoSuchHostedZone" not in str(type(e)):
            raise
    # For some reason TXT records need to be
    # manually quoted.
    seen = False
    for record in current_txt_records:
        for k, v in record.items():
            if '"{}"'.format(value) == v:
                seen = True
    if not seen:
        current_txt_records.append({"Value": '"{}"'.format(value)})

    if action == "DELETE" and len(current_txt_records) > 1:
        # If we want to delete one record out of many, we'll update the record to not include the deleted value instead.
        # This allows us to support concurrent issuance.
        current_txt_records = [
            record
            for record in current_txt_records
            if not (record.get("Value") == '"{}"'.format(value))
        ]
        action = "UPSERT"

    response = client.change_resource_record_sets(
        HostedZoneId=zone_id,
        ChangeBatch={
            "Changes": [
                {
                    "Action": action,
                    "ResourceRecordSet": {
                        "Name": domain,
                        "Type": "TXT",
                        "TTL": 300,
                        "ResourceRecords": current_txt_records,
                    },
                }
            ]
        },
    )
    return response["ChangeInfo"]["Id"]