34 lines
1.3 KiB
Python
34 lines
1.3 KiB
Python
|
from typing import Any, Optional, Awaitable, Callable, TypeVar
|
||
|
|
|
||
|
|
Handler = TypeVar("Handler", bound=Callable[[str, dict[str, Any]], Any])
|
||
|
|
ClientHandler = TypeVar("ClientHandler", bound=Callable[[dict[str, Any]], Any])
|
||
|
|
|
||
|
|
class _session_context_manager:
|
||
|
|
async def __aenter__(self) -> dict[str, Any]: ...
|
||
|
|
async def __aexit__(self, *args: list[Any]) -> None: ...
|
||
|
|
|
||
|
|
class AsyncServer:
|
||
|
|
def __init__(
|
||
|
|
self, cors_allowed_origins: str, logger: bool, engineio_logger: bool
|
||
|
|
): ...
|
||
|
|
async def emit(
|
||
|
|
self,
|
||
|
|
message: str,
|
||
|
|
body: Optional[dict[str, Any]] = None,
|
||
|
|
room: Optional[str] = None,
|
||
|
|
) -> None: ...
|
||
|
|
def session(self, sid: str) -> _session_context_manager: ...
|
||
|
|
def on(self, event: str) -> Callable[[Handler], Handler]: ...
|
||
|
|
def enter_room(self, sid: str, room: str) -> None: ...
|
||
|
|
def leave_room(self, sid: str, room: str) -> None: ...
|
||
|
|
def attach(self, app: Any) -> None: ...
|
||
|
|
|
||
|
|
class AsyncClient:
|
||
|
|
def on(self, event: str) -> Callable[[ClientHandler], ClientHandler]: ...
|
||
|
|
async def wait(self) -> None: ...
|
||
|
|
async def connect(self, server: str) -> None: ...
|
||
|
|
async def disconnect(self) -> None: ...
|
||
|
|
async def emit(
|
||
|
|
self, message: str, data: Optional[dict[str, Any]] = None
|
||
|
|
) -> None: ...
|