return {"msg": "Hello World"}
-@app.websocket_route("/ws")
+@app.websocket("/ws")
async def websocket(websocket: WebSocket):
await websocket.accept()
await websocket.send_json({"msg": "Hello WebSocket"})
from starlette.datastructures import State
from starlette.exceptions import HTTPException
from starlette.middleware import Middleware
+from starlette.middleware.base import BaseHTTPMiddleware
from starlette.middleware.errors import ServerErrorMiddleware
from starlette.middleware.exceptions import ExceptionMiddleware
from starlette.requests import Request
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
)
+
+ def websocket_route(
+ self, path: str, name: Union[str, None] = None
+ ) -> Callable[[DecoratedCallable], DecoratedCallable]:
+ def decorator(func: DecoratedCallable) -> DecoratedCallable:
+ self.router.add_websocket_route(path, func, name=name)
+ return func
+
+ return decorator
+
+ def on_event(
+ self, event_type: str
+ ) -> Callable[[DecoratedCallable], DecoratedCallable]:
+ return self.router.on_event(event_type)
+
+ def middleware(
+ self, middleware_type: str
+ ) -> Callable[[DecoratedCallable], DecoratedCallable]:
+ def decorator(func: DecoratedCallable) -> DecoratedCallable:
+ self.add_middleware(BaseHTTPMiddleware, dispatch=func)
+ return func
+
+ return decorator
+
+ def exception_handler(
+ self, exc_class_or_status_code: Union[int, Type[Exception]]
+ ) -> Callable[[DecoratedCallable], DecoratedCallable]:
+ def decorator(func: DecoratedCallable) -> DecoratedCallable:
+ self.add_exception_handler(exc_class_or_status_code, func)
+ return func
+
+ return decorator
self.default_response_class = default_response_class
self.generate_unique_id_function = generate_unique_id_function
+ def route(
+ self,
+ path: str,
+ methods: Optional[List[str]] = None,
+ name: Optional[str] = None,
+ include_in_schema: bool = True,
+ ) -> Callable[[DecoratedCallable], DecoratedCallable]:
+ def decorator(func: DecoratedCallable) -> DecoratedCallable:
+ self.add_route(
+ path,
+ func,
+ methods=methods,
+ name=name,
+ include_in_schema=include_in_schema,
+ )
+ return func
+
+ return decorator
+
def add_api_route(
self,
path: str,
return decorator
+ def websocket_route(
+ self, path: str, name: Union[str, None] = None
+ ) -> Callable[[DecoratedCallable], DecoratedCallable]:
+ def decorator(func: DecoratedCallable) -> DecoratedCallable:
+ self.add_websocket_route(path, func, name=name)
+ return func
+
+ return decorator
+
def include_router(
self,
router: "APIRouter",
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
)
+
+ def on_event(
+ self, event_type: str
+ ) -> Callable[[DecoratedCallable], DecoratedCallable]:
+ def decorator(func: DecoratedCallable) -> DecoratedCallable:
+ self.add_event_handler(event_type, func)
+ return func
+
+ return decorator
"Topic :: Internet :: WWW/HTTP",
]
dependencies = [
- "starlette==0.22.0",
+ "starlette>=0.22.0,<=0.23.0",
"pydantic >=1.6.2,!=1.7,!=1.7.1,!=1.7.2,!=1.7.3,!=1.8,!=1.8.1,<2.0.0",
]
dynamic = ["version"]
def test_websocket_invalid_path_doesnt_match():
with pytest.raises(WebSocketDisconnect):
- with client.websocket_connect("/itemsx/portal-gun") as websocket:
- websocket.receive_json()
+ with client.websocket_connect("/itemsx/portal-gun"):
+ pass