Method parse

src/websockets/http11.py:211–287

Parse a WebSocket handshake response. This is a generator-based coroutine. The reason phrase and headers are expected to contain only ASCII characters. Other characters are represented with surrogate escapes. Args: read_line: Generator-based co

(
        cls,
        read_line: Callable[[int], Generator[None, None, bytes | bytearray]],
        read_exact: Callable[[int], Generator[None, None, bytes | bytearray]],
        read_to_eof: Callable[[int], Generator[None, None, bytes | bytearray]],
        proxy: bool = False,
    )

    @classmethod
    def parse(
        cls,
        read_line: Callable[[int], Generator[None, None, bytes | bytearray]],
        read_exact: Callable[[int], Generator[None, None, bytes | bytearray]],
        read_to_eof: Callable[[int], Generator[None, None, bytes | bytearray]],
        proxy: bool = False,
    ) -> Generator[None, None, Response]:
        """
        Parse a WebSocket handshake response.

        This is a generator-based coroutine.

        The reason phrase and headers are expected to contain only ASCII
        characters. Other characters are represented with surrogate escapes.

        Args:
            read_line: Generator-based coroutine that reads a LF-terminated
                line or raises an exception if there isn't enough data.
            read_exact: Generator-based coroutine that reads the requested
                bytes or raises an exception if there isn't enough data.
            read_to_eof: Generator-based coroutine that reads until the end
                of the stream.

        Raises:
            EOFError: If the connection is closed without a full HTTP response.
            SecurityError: If the response exceeds a security limit.
            LookupError: If the response isn't well formatted.
            ValueError: If the response isn't well formatted.

        """
        # https://datatracker.ietf.org/doc/html/rfc7230#section-3.1.2

        try:
            status_line = yield from parse_line(read_line)
        except EOFError as exc:
            raise EOFError("connection closed while reading HTTP status line") from exc

        try:
            protocol, raw_status_code, raw_reason = status_line.split(b" ", 2)
        except ValueError:  # not enough values to unpack (expected 3, got 1-2)
            raise ValueError(f"invalid HTTP status line: {d(status_line)}") from None
        if proxy:  # some proxies still use HTTP/1.0
            if protocol not in [b"HTTP/1.1", b"HTTP/1.0"]:
                raise ValueError(
                    f"unsupported protocol; expected HTTP/1.1 or HTTP/1.0: "
                    f"{d(status_line)}"
                )
        else:
            if protocol != b"HTTP/1.1":
                raise ValueError(
                    f"unsupported protocol; expected HTTP/1.1: {d(status_line)}"
                )
        try:
            status_code = int(raw_status_code)
        except ValueError:  # invalid literal for int() with base 10
            raise ValueError(
                f"invalid status code; expected integer; got {d(raw_status_code)}"
            ) from None