|
2 | 2 | from collections.abc import Awaitable, Callable |
3 | 3 | from functools import wraps |
4 | 4 | from types import ModuleType |
5 | | -from typing import Any, Final |
| 5 | +from typing import Any, Final, TypeAlias |
6 | 6 |
|
7 | 7 | from aiohttp import web |
8 | 8 | from socketio import AsyncServer # type: ignore[import-untyped] |
9 | 9 |
|
10 | | -APP_CLIENT_SOCKET_DECORATED_HANDLERS_APPKEY: Final = web.AppKey( |
| 10 | +_CLIENT_SOCKET_DECORATED_HANDLERS_APPKEY: Final = web.AppKey( |
11 | 11 | "APP_CLIENT_SOCKET_DECORATED_HANDLERS", list[Callable] |
12 | 12 | ) |
13 | | -APP_CLIENT_SOCKET_SERVER_APPKEY: Final = web.AppKey( |
| 13 | +CLIENT_SOCKET_SERVER_APPKEY: Final = web.AppKey( |
14 | 14 | # NOTE: AsyncServer stub library is missing |
15 | 15 | "APP_CLIENT_SOCKET_SERVER", |
16 | 16 | AsyncServer, # type: ignore[var-annotated] |
17 | 17 | ) |
18 | 18 |
|
19 | 19 |
|
20 | 20 | def get_socket_server(app: web.Application) -> AsyncServer: |
21 | | - return app[APP_CLIENT_SOCKET_SERVER_APPKEY] |
| 21 | + return app[CLIENT_SOCKET_SERVER_APPKEY] |
22 | 22 |
|
23 | 23 |
|
24 | 24 | # The socket ID that was assigned to the client |
25 | | -SocketID = str |
| 25 | +SocketID: TypeAlias = str |
26 | 26 |
|
27 | 27 | # The environ argument is a dictionary in standard WSGI format containing the request information, including HTTP headers |
28 | | -EnvironDict = dict[str, Any] |
| 28 | +EnvironDict: TypeAlias = dict[str, Any] |
29 | 29 |
|
30 | 30 | # Connect event |
31 | | -SocketioConnectEventHandler = Callable[ |
| 31 | +SocketioConnectEventHandler: TypeAlias = Callable[ |
32 | 32 | [SocketID, EnvironDict, web.Application], Awaitable[None] |
33 | 33 | ] |
34 | 34 |
|
35 | 35 | # Disconnect event |
36 | | -SocketioDisconnectEventHandler = Callable[[SocketID, web.Application], Awaitable[None]] |
| 36 | +SocketioDisconnectEventHandler: TypeAlias = Callable[ |
| 37 | + [SocketID, web.Application], Awaitable[None] |
| 38 | +] |
37 | 39 |
|
38 | 40 | # Event |
39 | | -AnyData = Any |
40 | | -SocketioEventHandler = Callable[[SocketID, AnyData, web.Application], Awaitable[None]] |
| 41 | +AnyData: TypeAlias = Any |
| 42 | +SocketioEventHandler: TypeAlias = Callable[ |
| 43 | + [SocketID, AnyData, web.Application], Awaitable[None] |
| 44 | +] |
41 | 45 |
|
42 | 46 | _socketio_handlers_registry: list[ |
43 | 47 | ( |
@@ -82,7 +86,7 @@ def register_socketio_handlers(app: web.Application, module: ModuleType): |
82 | 86 | partial_fcts = [ |
83 | 87 | _socket_io_handler(app)(func_handler) for func_handler in member_fcts |
84 | 88 | ] |
85 | | - app[APP_CLIENT_SOCKET_DECORATED_HANDLERS_APPKEY] = partial_fcts |
| 89 | + app[_CLIENT_SOCKET_DECORATED_HANDLERS_APPKEY] = partial_fcts |
86 | 90 |
|
87 | 91 | # register the fcts |
88 | 92 | for func in partial_fcts: |
|
0 commit comments