| 1160 | |
| 1161 | |
| 1162 | def remove_path_argument( |
| 1163 | ws_handler: ( |
| 1164 | Callable[[WebSocketServerProtocol], Awaitable[Any]] |
| 1165 | | Callable[[WebSocketServerProtocol, str], Awaitable[Any]] |
| 1166 | ), |
| 1167 | ) -> Callable[[WebSocketServerProtocol], Awaitable[Any]]: |
| 1168 | try: |
| 1169 | inspect.signature(ws_handler).bind(None) |
| 1170 | except TypeError: |
| 1171 | try: |
| 1172 | inspect.signature(ws_handler).bind(None, "") |
| 1173 | except TypeError: # pragma: no cover |
| 1174 | # ws_handler accepts neither one nor two arguments; leave it alone. |
| 1175 | pass |
| 1176 | else: |
| 1177 | # ws_handler accepts two arguments; activate backwards compatibility. |
| 1178 | warnings.warn("remove second argument of ws_handler", DeprecationWarning) |
| 1179 | |
| 1180 | async def _ws_handler(websocket: WebSocketServerProtocol) -> Any: |
| 1181 | return await cast( |
| 1182 | Callable[[WebSocketServerProtocol, str], Awaitable[Any]], |
| 1183 | ws_handler, |
| 1184 | )(websocket, websocket.path) |
| 1185 | |
| 1186 | return _ws_handler |
| 1187 | |
| 1188 | return cast( |
| 1189 | Callable[[WebSocketServerProtocol], Awaitable[Any]], |
| 1190 | ws_handler, |
| 1191 | ) |