Function tcp

dns/query.py:1204–1271

Return the response obtained after sending a query via TCP. *q*, a ``dns.message.Message``, the query to send *where*, a ``str`` containing an IPv4 or IPv6 address, where to send the message. *timeout*, a ``float`` or ``None``, the number of seconds to wait before the query ti

(
    q: dns.message.Message,
    where: str,
    timeout: float | None = None,
    port: int = 53,
    source: str | None = None,
    source_port: int = 0,
    one_rr_per_rrset: bool = False,
    ignore_trailing: bool = False,
    sock: Any | None = None,
)

def tcp(
    q: dns.message.Message,
    where: str,
    timeout: float | None = None,
    port: int = 53,
    source: str | None = None,
    source_port: int = 0,
    one_rr_per_rrset: bool = False,
    ignore_trailing: bool = False,
    sock: Any | None = None,
) -> dns.message.Message:
    """Return the response obtained after sending a query via TCP.

    *q*, a ``dns.message.Message``, the query to send

    *where*, a ``str`` containing an IPv4 or IPv6 address, where
    to send the message.

    *timeout*, a ``float`` or ``None``, the number of seconds to wait before the
    query times out.  If ``None``, the default, wait forever.

    *port*, an ``int``, the port send the message to.  The default is 53.

    *source*, a ``str`` containing an IPv4 or IPv6 address, specifying
    the source address.  The default is the wildcard address.

    *source_port*, an ``int``, the port from which to send the message.
    The default is 0.

    *one_rr_per_rrset*, a ``bool``.  If ``True``, put each RR into its own
    RRset.

    *ignore_trailing*, a ``bool``.  If ``True``, ignore trailing
    junk at end of the received message.

    *sock*, a ``socket.socket``, or ``None``, the connected socket to use for the
    query.  If ``None``, the default, a socket is created.  Note that
    if a socket is provided, it must be a nonblocking connected stream
    socket, and *where*, *port*, *source* and *source_port* are ignored.

    Returns a ``dns.message.Message``.
    """

    wire = q.to_wire()
    (begin_time, expiration) = _compute_times(timeout)
    if sock:
        cm: contextlib.AbstractContextManager = contextlib.nullcontext(sock)
    else:
        (af, destination, source) = _destination_and_source(
            where, port, source, source_port, True
        )
        assert af is not None
        cm = make_socket(af, socket.SOCK_STREAM, source)
    with cm as s:
        if not sock:
            # pylint: disable=possibly-used-before-assignment
            _connect(s, destination, expiration)  # pyright: ignore
        send_tcp(s, wire, expiration)