ms_txt_verification.py

from otto.clients.gandi_client import GandiClient, GandiDomainRecord
from argparse import ArgumentParser
import ujson
import requests
from tenacity import retry, retry_if_exception, stop_after_attempt, RetryCallState, wait_random

gandi = GandiClient(api_key="")

# Added a method to the gandi client

#    def add_single_domain_record(self, domain: str, record: GandiDomainRecord):
#        return self.http.put(
#           f"{self._base_dns_endpoint}/{domain}/records/{record.rrset_name}/{record.rrset_type}", json=record.dict()
#        )




class retry_if_status_code(retry_if_exception):
    def __init__(self, *status_codes: int) -> None:
        self.status_codes = status_codes

    def __call__(self, retry_state: RetryCallState) -> bool:
        if retry_state.outcome is None:
            raise RuntimeError("__call__() called before outcome was set")

        exception = retry_state.outcome.exception()
        return isinstance(exception, requests.HTTPError) and exception.response.status_code in self.status_codes


@retry(retry=retry_if_status_code(429, 500), reraise=True, stop=stop_after_attempt(10), wait=wait_random(10))
def main(domain, record):
    message = {"domain": domain, "error": None}
    try:
        resp = gandi.add_single_domain_record(
            domain,
            GandiDomainRecord(
                rrset_name="@",
                rrset_type="TXT",
                rrset_ttl=3600,
                rrset_values=[record],
            ),
        )
        resp.raise_for_status()
    except requests.HTTPError as e:
        message["error"] = str(e)
    finally:
        print(ujson.dumps(message))


if __name__ == "__main__":
    parser = ArgumentParser()
    parser.add_argument("domain", help="Domain without www prefix")
    parser.add_argument("record", help="TXT record value")
    args = parser.parse_args()

    main(args.domain, args.record)