Return the response obtained after sending a query via UDP. *sock*, a ``dns.asyncbackend.DatagramSocket``, or ``None``, the socket to use for the query. If ``None``, the default, a socket is created. Note that if a socket is provided, the *source*, *source_port*, and *backend* are
(
q: dns.message.Message,
where: str,
timeout: float | None = None,
port: int = 53,
source: str | None = None,
source_port: int = 0,
ignore_unexpected: bool = False,
one_rr_per_rrset: bool = False,
ignore_trailing: bool = False,
raise_on_truncation: bool = False,
sock: dns.asyncbackend.DatagramSocket | None = None,
backend: dns.asyncbackend.Backend | None = None,
ignore_errors: bool = False,
)
| 180 | |
| 181 | |
| 182 | async def udp( |
| 183 | q: dns.message.Message, |
| 184 | where: str, |
| 185 | timeout: float | None = None, |
| 186 | port: int = 53, |
| 187 | source: str | None = None, |
| 188 | source_port: int = 0, |
| 189 | ignore_unexpected: bool = False, |
| 190 | one_rr_per_rrset: bool = False, |
| 191 | ignore_trailing: bool = False, |
| 192 | raise_on_truncation: bool = False, |
| 193 | sock: dns.asyncbackend.DatagramSocket | None = None, |
| 194 | backend: dns.asyncbackend.Backend | None = None, |
| 195 | ignore_errors: bool = False, |
| 196 | ) -> dns.message.Message: |
| 197 | """Return the response obtained after sending a query via UDP. |
| 198 | |
| 199 | *sock*, a ``dns.asyncbackend.DatagramSocket``, or ``None``, |
| 200 | the socket to use for the query. If ``None``, the default, a |
| 201 | socket is created. Note that if a socket is provided, the |
| 202 | *source*, *source_port*, and *backend* are ignored. |
| 203 | |
| 204 | *backend*, a ``dns.asyncbackend.Backend``, or ``None``. If ``None``, |
| 205 | the default, then dnspython will use the default backend. |
| 206 | |
| 207 | See :py:func:`dns.query.udp()` for the documentation of the other |
| 208 | parameters, exceptions, and return type of this method. |
| 209 | """ |
| 210 | wire = q.to_wire() |
| 211 | (begin_time, expiration) = _compute_times(timeout) |
| 212 | af = dns.inet.af_for_address(where) |
| 213 | destination = _lltuple((where, port), af) |
| 214 | if sock: |
| 215 | cm: contextlib.AbstractAsyncContextManager = NullContext(sock) |
| 216 | else: |
| 217 | if not backend: |
| 218 | backend = dns.asyncbackend.get_default_backend() |
| 219 | stuple = _source_tuple(af, source, source_port) |
| 220 | if backend.datagram_connection_required(): |
| 221 | dtuple = (where, port) |
| 222 | else: |
| 223 | dtuple = None |
| 224 | cm = await backend.make_socket(af, socket.SOCK_DGRAM, 0, stuple, dtuple) |
| 225 | async with cm as s: |
| 226 | await send_udp(s, wire, destination, expiration) # pyright: ignore |
| 227 | (r, received_time, _) = await receive_udp( |
| 228 | s, # pyright: ignore |
| 229 | destination, |
| 230 | expiration, |
| 231 | ignore_unexpected, |
| 232 | one_rr_per_rrset, |
| 233 | q.keyring, |
| 234 | q.mac, |
| 235 | ignore_trailing, |
| 236 | raise_on_truncation, |
| 237 | ignore_errors, |
| 238 | q, |
| 239 | ) |
no test coverage detected