Add stubs for Flask-SocketIO

pyrightconfig.stricter.json

@@ -35,6 +35,7 @@
         "stubs/dateparser",
         "stubs/docutils",
         "stubs/Flask-Migrate",
+        "stubs/Flask-SocketIO",
         "stubs/fpdf2",
         "stubs/google-cloud-ndb",
         "stubs/html5lib",

stubs/Flask-SocketIO/@tests/stubtest_allowlist.txt

@@ -0,0 +1,4 @@
+# private attributes / methods, not present in docs
+flask_socketio.test_client.SocketIOTestClient.clients
+flask_socketio.gevent_socketio_found
+flask_socketio.call

stubs/Flask-SocketIO/METADATA.toml

@@ -0,0 +1,3 @@
+version = "5.3.*"
+requires = ["Flask>=0.9"]
+upstream_repository = "https://github.com/miguelgrinberg/flask-socketio"

stubs/Flask-SocketIO/flask_socketio/__init__.pyi

@@ -0,0 +1,133 @@
+from _typeshed import Incomplete
+from collections.abc import Callable
+from threading import Thread
+from typing import Any, Protocol, TypeVar, overload
+from typing_extensions import ParamSpec, TypeAlias
+
+from flask import Flask
+from flask.testing import FlaskClient
+
+from .namespace import Namespace
+from .test_client import SocketIOTestClient
+
+_P = ParamSpec("_P")
+_R_co = TypeVar("_R_co", covariant=True)
+_ExceptionHandler: TypeAlias = Callable[[BaseException], _R_co]
+_Handler: TypeAlias = Callable[_P, _R_co]
+
+class _HandlerDecorator(Protocol):
+    def __call__(self, handler: _Handler[_P, _R_co]) -> _Handler[_P, _R_co]: ...
+
+class _ExceptionHandlerDecorator(Protocol):
+    def __call__(self, exception_handler: _ExceptionHandler[_R_co]) -> _ExceptionHandler[_R_co]: ...
+
+class SocketIO:
+    # Many instance attributes are deliberately not included here,
+    # as the maintainer of Flask-SocketIO considers them private, internal details:
+    # https://github.com/python/typeshed/pull/10735#discussion_r1330768869
+    def __init__(
+        self,
+        app: Flask | None = None,
+        *,
+        # SocketIO options
+        manage_session: bool = True,
+        message_queue: str | None = None,
+        channel: str = "flask-socketio",
+        path: str = "socket.io",
+        resource: str = "socket.io",
+        **kwargs,  # TODO: Socket.IO server options, Engine.IO server config
+    ) -> None: ...
+    def init_app(
+        self,
+        app: Flask,
+        *,
+        # SocketIO options
+        manage_session: bool = True,
+        message_queue: str | None = None,
+        channel: str = "flask-socketio",
+        path: str = "socket.io",
+        resource: str = "socket.io",
+        **kwargs,  # TODO: Socket.IO server options, Engine.IO server config: ...
+    ) -> None: ...
+    def on(self, message: str, namespace: str | None = None) -> _HandlerDecorator: ...
+    def on_error(self, namespace: str | None = None) -> _ExceptionHandlerDecorator: ...
+    def on_error_default(self, exception_handler: _ExceptionHandler[_R_co]) -> _ExceptionHandler[_R_co]: ...
+    def on_event(self, message: str, handler: _Handler[[Incomplete], object], namespace: str | None = None) -> None: ...
+    @overload
+    def event(self, __event_handler: _Handler[_P, _R_co]) -> _Handler[_P, _R_co]: ...
+    @overload
+    def event(self, namespace: str | None = None, *args, **kwargs) -> _HandlerDecorator: ...
+    def on_namespace(self, namespace_handler: Namespace) -> None: ...
+    def emit(
+        self,
+        event: str,
+        *args,
+        namespace: str = "/",  # / is the default (global) namespace
+        to: str | None = None,
+        include_self: bool = True,
+        skip_sid: str | list[str] | None = None,
+        callback: Callable[..., Incomplete] | None = None,
+    ) -> None: ...
+    def call(
+        self,
+        event: str,
+        *args,
+        namespace: str = "/",  # / is the default (global) namespace
+        to: str | None = None,
+        timeout: int = 60,  # seconds
+        ignore_queue: bool = False,
+    ): ...
+    def send(
+        self,
+        data: Any,
+        json: bool = False,
+        namespace: str | None = None,
+        to: str | None = None,
+        callback: Callable[..., Incomplete] | None = None,
+        include_self: bool = True,
+        skip_sid: list[str] | str | None = None,
+        **kwargs,
+    ) -> None: ...
+    def close_room(self, room: str, namespace: str | None = None) -> None: ...
+    def run(
+        self,
+        app,
+        host: str | None = None,
+        port: int | None = None,
+        *,
+        debug: bool = True,
+        use_reloader: bool,
+        reloader_options: dict[str, Incomplete] = {},
+        log_output: bool,
+        allow_unsafe_werkzeug: bool = False,
+        **kwargs,
+    ) -> None: ...
+    def stop(self) -> None: ...
+    def start_background_task(self, target: Callable[_P, None], *args: _P.args, **kwargs: _P.kwargs) -> Thread: ...
+    def sleep(self, seconds: int = 0): ...
+    def test_client(
+        self,
+        app: Flask,
+        namespace: str | None = None,
+        query_string: str | None = None,
+        headers: dict[str, Incomplete] | None = None,
+        auth: dict[str, Incomplete] | None = None,
+        flask_test_client: FlaskClient | None = None,
+    ) -> SocketIOTestClient: ...
+
+def emit(
+    event,
+    *args,
+    namespace: str = "/",  # / is the default (global) namespace
+    to: str | None = None,
+    include_self: bool = True,
+    skip_sid: str | list[str] | None = None,
+    callback: Callable[..., Incomplete] | None = None,
+    broadcast: bool = False,
+) -> None: ...
+def send(message: str, **kwargs) -> None: ...
+def join_room(room, sid: str | None = None, namespace: str | None = None) -> None: ...
+def leave_room(room, sid: str | None = None, namespace: str | None = None) -> None: ...
+def close_room(room, namespace: str | None = None) -> None: ...
+def rooms(sid: str | None = None, namespace: str | None = None) -> list[str]: ...
+def disconnect(sid: str | None = None, namespace: str | None = None, silent: bool = False) -> None: ...

stubs/Flask-SocketIO/flask_socketio/namespace.pyi

@@ -0,0 +1,74 @@
+from _typeshed import Incomplete
+from collections.abc import Callable
+from typing import Any, Protocol, TypeVar
+
+_T = TypeVar("_T")
+
+# at runtime, socketio.namespace.BaseNamespace, but socketio isn't py.typed
+class _BaseNamespace(Protocol):
+    def is_asyncio_based(self) -> bool: ...
+    def trigger_event(self, event: str, *args): ...
+
+# at runtime, socketio.namespace.BaseNamespace, but socketio isn't py.typed
+class _Namespace(_BaseNamespace, Protocol):
+    def emit(
+        self,
+        event: str,
+        data: Incomplete | None = None,
+        to=None,
+        room: str | None = None,
+        skip_sid=None,
+        namespace: str | None = None,
+        callback: Callable[..., Incomplete] | None = None,
+        ignore_queue: bool = False,
+    ): ...
+    def send(
+        self,
+        data: Incomplete,
+        to=None,
+        room: str | None = None,
+        skip_sid=None,
+        namespace: str | None = None,
+        callback: Callable[..., Incomplete] | None = None,
+        ignore_queue: bool = False,
+    ) -> None: ...
+    def call(
+        self,
+        event: str,
+        data: Incomplete | None = None,
+        to=None,
+        sid=None,
+        namespace: str | None = None,
+        timeout=None,
+        ignore_queue: bool = False,
+    ): ...
+    def enter_room(self, sid, room: str, namespace: str | None = None): ...
+    def leave_room(self, sid, room: str, namespace: str | None = None): ...
+    def close_room(self, room: str, namespace: str | None = None): ...
+    def rooms(self, sid, namespace: str | None = None): ...
+    def get_session(self, sid, namespace: str | None = None): ...
+    def save_session(self, sid, session, namespace: str | None = None): ...
+    def session(self, sid, namespace: str | None = None): ...
+    def disconnect(self, sid, namespace: str | None = None): ...
+
+class Namespace(_Namespace):
+    def __init__(self, namespace: str | None = None) -> None: ...
+    def trigger_event(self, event: str, *args): ...
+    def emit(  # type: ignore[override]
+        self,
+        event: str,
+        data: Incomplete | None = None,
+        room: str | None = None,
+        include_self: bool = True,
+        namespace: str | None = None,
+        callback: Callable[..., _T] | None = None,
+    ) -> _T | tuple[str, int]: ...
+    def send(  # type: ignore[override]
+        self,
+        data: Incomplete,
+        room: str | None = None,
+        include_self: bool = True,
+        namespace: str | None = None,
+        callback: Callable[..., Any] | None = None,
+    ) -> None: ...
+    def close_room(self, room: str, namespace: str | None = None) -> None: ...

stubs/Flask-SocketIO/flask_socketio/test_client.pyi

@@ -0,0 +1,41 @@
+from _typeshed import Incomplete
+from typing import Any
+from typing_extensions import TypedDict
+
+from flask import Flask
+from flask.testing import FlaskClient
+
+class _Packet(TypedDict):
+    name: str
+    args: Any
+    namespace: str
+
+class SocketIOTestClient:
+    def __init__(
+        self,
+        app: Flask,
+        socketio,
+        namespace: str | None = None,
+        query_string: str | None = None,
+        headers: dict[str, Incomplete] | None = None,
+        auth: dict[str, Incomplete] | None = None,
+        flask_test_client: FlaskClient | None = None,
+    ) -> None: ...
+    def is_connected(self, namespace: str | None = None) -> bool: ...
+    def connect(
+        self,
+        namespace: str | None = None,
+        query_string: str | None = None,
+        headers: dict[str, Incomplete] | None = None,
+        auth: dict[str, Incomplete] | None = None,
+    ) -> None: ...
+    def disconnect(self, namespace: str | None = None) -> None: ...
+    def emit(self, event: str, *args, callback: bool = True, namespace: str | None = None) -> Incomplete | None: ...
+    def send(
+        self,
+        data: str | dict[str, Incomplete] | list[Incomplete],
+        json: bool = False,
+        callback: bool = False,
+        namespace: str | None = None,
+    ): ...
+    def get_received(self, namespace: str | None = None) -> list[_Packet]: ...