Method next_request

dns/resolver.py:685–758

Get the next request to send, and check the cache. Returns a (request, answer) tuple. At most one of request or answer will not be None.

(
        self,
    )

        self.backoff = 0.0

    def next_request(
        self,
    ) -> Tuple[dns.message.QueryMessage | None, Answer | None]:
        """Get the next request to send, and check the cache.

        Returns a (request, answer) tuple.  At most one of request or
        answer will not be None.
        """

        # We return a tuple instead of Union[Message,Answer] as it lets
        # the caller avoid isinstance().

        while len(self.qnames) > 0:
            self.qname = self.qnames.pop(0)

            # Do we know the answer?
            if self.resolver.cache:
                answer = self.resolver.cache.get(
                    (self.qname, self.rdtype, self.rdclass)
                )
                if answer is not None:
                    if answer.rrset is None and self.raise_on_no_answer:
                        raise NoAnswer(response=answer.response)
                    else:
                        return (None, answer)
                answer = self.resolver.cache.get(
                    (self.qname, dns.rdatatype.ANY, self.rdclass)
                )
                if answer is not None and answer.response.rcode() == dns.rcode.NXDOMAIN:
                    # cached NXDOMAIN; record it and continue to next
                    # name.
                    self.nxdomain_responses[self.qname] = answer.response
                    continue

            # Build the request
            request = dns.message.make_query(self.qname, self.rdtype, self.rdclass)
            if self.resolver.keyname is not None:
                request.use_tsig(
                    self.resolver.keyring,
                    self.resolver.keyname,
                    algorithm=self.resolver.keyalgorithm,
                )
            request.use_edns(
                self.resolver.edns,
                self.resolver.ednsflags,
                self.resolver.payload,
                options=self.resolver.ednsoptions,
            )
            if self.resolver.flags is not None:
                request.flags = self.resolver.flags

            self.nameservers = self.resolver._enrich_nameservers(
                self.resolver._nameservers,
                self.resolver.nameserver_ports,
                self.resolver.port,
            )
            if self.resolver.rotate:
                random.shuffle(self.nameservers)