Asynchronous DNS stub resolver.
| 43 | |
| 44 | |
| 45 | class Resolver(dns.resolver.BaseResolver): |
| 46 | """Asynchronous DNS stub resolver.""" |
| 47 | |
| 48 | async def resolve( |
| 49 | self, |
| 50 | qname: dns.name.Name | str, |
| 51 | rdtype: dns.rdatatype.RdataType | str = dns.rdatatype.A, |
| 52 | rdclass: dns.rdataclass.RdataClass | str = dns.rdataclass.IN, |
| 53 | tcp: bool = False, |
| 54 | source: str | None = None, |
| 55 | raise_on_no_answer: bool = True, |
| 56 | source_port: int = 0, |
| 57 | lifetime: float | None = None, |
| 58 | search: bool | None = None, |
| 59 | backend: dns.asyncbackend.Backend | None = None, |
| 60 | ) -> dns.resolver.Answer: |
| 61 | """Query nameservers asynchronously to find the answer to the question. |
| 62 | |
| 63 | *backend*, a ``dns.asyncbackend.Backend``, or ``None``. If ``None``, |
| 64 | the default, then dnspython will use the default backend. |
| 65 | |
| 66 | See :py:func:`dns.resolver.Resolver.resolve()` for the |
| 67 | documentation of the other parameters, exceptions, and return |
| 68 | type of this method. |
| 69 | """ |
| 70 | |
| 71 | resolution = dns.resolver._Resolution( |
| 72 | self, qname, rdtype, rdclass, tcp, raise_on_no_answer, search |
| 73 | ) |
| 74 | if not backend: |
| 75 | backend = dns.asyncbackend.get_default_backend() |
| 76 | start = time.time() |
| 77 | while True: |
| 78 | (request, answer) = resolution.next_request() |
| 79 | # Note we need to say "if answer is not None" and not just |
| 80 | # "if answer" because answer implements __len__, and python |
| 81 | # will call that. We want to return if we have an answer |
| 82 | # object, including in cases where its length is 0. |
| 83 | if answer is not None: |
| 84 | # cache hit! |
| 85 | return answer |
| 86 | assert request is not None # needed for type checking |
| 87 | done = False |
| 88 | while not done: |
| 89 | (nameserver, tcp, backoff) = resolution.next_nameserver() |
| 90 | if backoff: |
| 91 | await backend.sleep(backoff) |
| 92 | timeout = self._compute_timeout(start, lifetime, resolution.errors) |
| 93 | try: |
| 94 | response = await nameserver.async_query( |
| 95 | request, |
| 96 | timeout=timeout, |
| 97 | source=source, |
| 98 | source_port=source_port, |
| 99 | max_size=tcp, |
| 100 | backend=backend, |
| 101 | ) |
| 102 | except Exception as ex: |
no outgoing calls
no test coverage detected
searching dependent graphs…