|
13 | 13 |
|
14 | 14 | import pydantic |
15 | 15 | from fastapi import APIRouter, Depends |
16 | | -from fastapi.routing import APIRoute |
| 16 | +from fastapi.routing import APIRoute, APIWebSocketRoute |
17 | 17 | from starlette.routing import Route, WebSocketRoute |
18 | 18 |
|
19 | 19 | PYDANTIC_VERSION = pydantic.VERSION |
@@ -108,12 +108,16 @@ def _register_endpoints(router: APIRouter, cls: Type[Any], *urls: str) -> None: |
108 | 108 | _allocate_routes_by_method_name(router, url, function_members) |
109 | 109 | router_roles = [] |
110 | 110 | for route in router.routes: |
111 | | - if not isinstance(route, APIRoute): |
112 | | - raise ValueError("The provided routes should be of type APIRoute") |
| 111 | + if not isinstance(route, APIRoute) and not isinstance(route, APIWebSocketRoute): |
| 112 | + raise ValueError("The provided routes should be of type APIRoute or APIWebSocketRoute") |
113 | 113 |
|
114 | | - route_methods: Any = route.methods |
115 | | - cast(Tuple[Any], route_methods) |
116 | | - router_roles.append((route.path, tuple(route_methods))) |
| 114 | + if isinstance(route, APIRoute): |
| 115 | + route_methods: Any = route.methods |
| 116 | + cast(Tuple[Any], route_methods) |
| 117 | + router_roles.append((route.path, tuple(route_methods))) |
| 118 | + |
| 119 | + if isinstance(route, APIWebSocketRoute): |
| 120 | + router_roles.append((route.path, tuple(["WS"]))) |
117 | 121 |
|
118 | 122 | if len(set(router_roles)) != len(router_roles): |
119 | 123 | raise Exception("An identical route role has been implemented more then once") |
|
0 commit comments