Class ServerProtocol

src/websockets/server.py:48–567

Sans-I/O implementation of a WebSocket server connection. Args: origins: Acceptable values of the ``Origin`` header. Values can be :class:`str` to test for an exact match or regular expressions compiled by :func:`re.compile` to test against a pattern. Includ

class ServerProtocol(Protocol):
    """
    Sans-I/O implementation of a WebSocket server connection.

    Args:
        origins: Acceptable values of the ``Origin`` header. Values can be
            :class:`str` to test for an exact match or regular expressions
            compiled by :func:`re.compile` to test against a pattern. Include
            :obj:`None` in the list if the lack of an origin is acceptable.
            This is useful for defending against Cross-Site WebSocket
            Hijacking attacks.
        extensions: List of supported extensions, in order in which they
            should be tried.
        subprotocols: List of supported subprotocols, in order of decreasing
            preference.
        select_subprotocol: Callback for selecting a subprotocol among
            those supported by the client and the server. It has the same
            signature as the :meth:`select_subprotocol` method, including a
            :class:`ServerProtocol` instance as first argument.
        state: Initial state of the WebSocket connection.
        max_size: Maximum size of incoming messages in bytes.
            :obj:`None` disables the limit. You may pass a ``(max_message_size,
            max_fragment_size)`` tuple to set different limits for messages and
            fragments when you expect long messages sent in short fragments.
        logger: Logger for this connection;
            defaults to ``logging.getLogger("websockets.server")``;
            see the :doc:`logging guide <../../topics/logging>` for details.

    """

    def __init__(
        self,
        *,
        origins: Sequence[Origin | re.Pattern[str] | None] | None = None,
        extensions: Sequence[ServerExtensionFactory] | None = None,
        subprotocols: Sequence[Subprotocol] | None = None,
        select_subprotocol: (
            Callable[
                [ServerProtocol, Sequence[Subprotocol]],
                Subprotocol | None,
            ]
            | None
        ) = None,
        state: State = CONNECTING,
        max_size: int | None | tuple[int | None, int | None] = 2**20,
        logger: LoggerLike | None = None,
    ) -> None:
        super().__init__(
            side=SERVER,
            state=state,
            max_size=max_size,
            logger=logger,
        )
        self.origins = origins
        self.available_extensions = extensions
        self.available_subprotocols = subprotocols
        if select_subprotocol is not None:
            # Bind select_subprotocol then shadow self.select_subprotocol.