patx/micropie
add websocket functionality
Commit 4db870a · patx · 2025-06-23T23:37:44-04:00
Comments
No comments yet.
Diff
diff --git a/MicroPie.py b/MicroPie.py
index 8ec5b0a..b26116b 100644
--- a/MicroPie.py
+++ b/MicroPie.py
@@ -12,7 +12,7 @@ License: BSD3 (see LICENSE for details)
"""
__author__ = 'Harrison Erd'
-__version__ = '0.12.2'
+__version__ = '0.13-dev'
__license__ = 'BSD3'
import asyncio
@@ -89,7 +89,7 @@ class InMemorySessionBackend(SessionBackend):
# -----------------------------
-# Request Object
+# Request Objects
# -----------------------------
current_request: contextvars.ContextVar[Any] = contextvars.ContextVar("current_request")
@@ -103,7 +103,7 @@ class Request:
scope: The ASGI scope dictionary for the request.
"""
self.scope: Dict[str, Any] = scope
- self.method: str = scope["method"]
+ self.method: str = scope.get("method", "")
self.path_params: List[str] = []
self.query_params: Dict[str, List[str]] = {}
self.body_params: Dict[str, List[str]] = {}
@@ -115,6 +115,141 @@ class Request:
for k, v in scope.get("headers", [])
}
+class WebSocketRequest(Request):
+ """Represents a WebSocket request in the MicroPie framework."""
+ def __init__(self, scope: Dict[str, Any]) -> None:
+ super().__init__(scope)
+
+class WebSocket:
+ """Manages WebSocket communication in the MicroPie framework."""
+ def __init__(self, receive: Callable[[], Awaitable[Dict[str, Any]]], send: Callable[[Dict[str, Any]], Awaitable[None]]) -> None:
+ """
+ Initialize a WebSocket instance.
+
+ Args:
+ receive: The ASGI receive callable.
+ send: The ASGI send callable.
+ """
+ self.receive = receive
+ self.send = send
+ self.accepted = False
+ self.session_id: Optional[str] = None
+
+ async def accept(self, subprotocol: Optional[str] = None, session_id: Optional[str] = None) -> None:
+ """
+ Accept the WebSocket connection.
+
+ Args:
+ subprotocol: Optional subprotocol to use.
+ session_id: Optional session ID to set in a cookie during the handshake.
+ """
+ if self.accepted:
+ raise RuntimeError("WebSocket connection already accepted")
+ # Handle initial connect event
+ message = await self.receive()
+ if message["type"] != "websocket.connect":
+ raise ValueError(f"Expected websocket.connect, got {message['type']}")
+ headers = []
+ if session_id:
+ headers.append(("Set-Cookie", f"session_id={session_id}; Path=/; SameSite=Lax; HttpOnly; Secure;"))
+ self.session_id = session_id
+ await self.send({
+ "type": "websocket.accept",
+ "subprotocol": subprotocol,
+ "headers": [(k.encode("latin-1"), v.encode("latin-1")) for k, v in headers]
+ })
+ self.accepted = True
+
+ async def receive_text(self) -> str:
+ """
+ Receive a text message from the WebSocket.
+
+ Returns:
+ The received text message.
+
+ Raises:
+ ConnectionClosed: If the connection is closed.
+ ValueError: If an unexpected message type is received.
+ """
+ message = await self.receive()
+ if message["type"] == "websocket.receive":
+ return message.get("text", message.get("bytes", b"").decode("utf-8", "ignore"))
+ elif message["type"] == "websocket.disconnect":
+ raise ConnectionClosed()
+ raise ValueError(f"Unexpected message type: {message['type']}")
+
+ async def receive_bytes(self) -> bytes:
+ """
+ Receive a binary message from the WebSocket.
+
+ Returns:
+ The received binary message.
+
+ Raises:
+ ConnectionClosed: If the connection is closed.
+ ValueError: If an unexpected message type is received.
+ """
+ message = await self.receive()
+ if message["type"] == "websocket.receive":
+ return message.get("bytes", b"") or message.get("text", "").encode("utf-8")
+ elif message["type"] == "websocket.disconnect":
+ raise ConnectionClosed()
+ raise ValueError(f"Unexpected message type: {message['type']}")
+
+ async def send_text(self, data: str) -> None:
+ """
+ Send a text message over the WebSocket.
+
+ Args:
+ data: The text message to send.
+
+ Raises:
+ RuntimeError: If the connection is not accepted.
+ """
+ if not self.accepted:
+ raise RuntimeError("WebSocket connection not accepted")
+ await self.send({
+ "type": "websocket.send",
+ "text": data
+ })
+
+ async def send_bytes(self, data: bytes) -> None:
+ """
+ Send a binary message over the WebSocket.
+
+ Args:
+ data: The binary message to send.
+
+ Raises:
+ RuntimeError: If the connection is not accepted.
+ """
+ if not self.accepted:
+ raise RuntimeError("WebSocket connection not accepted")
+ await self.send({
+ "type": "websocket.send",
+ "bytes": data
+ })
+
+ async def close(self, code: int = 1000, reason: Optional[str] = None) -> None:
+ """
+ Close the WebSocket connection.
+
+ Args:
+ code: The closure code (default: 1000).
+ reason: Optional reason for closure.
+ """
+ if self.accepted:
+ await self.send({
+ "type": "websocket.close",
+ "code": code,
+ "reason": reason or ""
+ })
+ self.accepted = False
+
+class ConnectionClosed(Exception):
+ """Raised when a WebSocket connection is closed."""
+ pass
+
# -----------------------------
# Middleware Abstraction
@@ -152,7 +287,7 @@ class HttpMiddleware(ABC):
# -----------------------------
class App:
"""
- ASGI application for handling HTTP requests in MicroPie.
+ ASGI application for handling HTTP and WebSocket requests in MicroPie.
It supports pluggable session backends via the 'session_backend' attribute
and pluggable middlewares via the 'middlewares' list.
"""
@@ -194,8 +329,10 @@ class App:
"""
if scope["type"] == "http":
await self._asgi_app_http(scope, receive, send)
+ elif scope["type"] == "websocket":
+ await self._asgi_app_websocket(scope, receive, send)
else:
- pass # Handle websockets, lifespan and more in the future.
+ pass # Handle lifespan and other scopes in the future.
async def _asgi_app_http(
self,
@@ -358,6 +495,92 @@ class App:
finally:
current_request.reset(token)
+ async def _asgi_app_websocket(
+ self,
+ scope: Dict[str, Any],
+ receive: Callable[[], Awaitable[Dict[str, Any]]],
+ send: Callable[[Dict[str, Any]], Awaitable[None]]
+ ) -> None:
+ """
+ ASGI application entry point for handling WebSocket requests.
+
+ Args:
+ scope: The ASGI scope dictionary.
+ receive: The callable to receive ASGI events.
+ send: The callable to send ASGI events.
+ """
+ request: WebSocketRequest = WebSocketRequest(scope)
+ token = current_request.set(request)
+ try:
+ # Parse request details (query params, cookies, session)
+ request.query_params = parse_qs(scope.get("query_string", b"").decode("utf-8", "ignore"))
+ cookies = self._parse_cookies(request.headers.get("cookie", ""))
+ request.session = await self.session_backend.load(cookies.get("session_id", "")) or {}
+
+ # Parse path and find handler
+ path: str = scope["path"].lstrip("/")
+ parts: List[str] = path.split("/") if path else []
+ func_name: str = parts[0] if parts else "ws_index"
+ if func_name.startswith("_"):
+ await self._send_websocket_close(send, 1008, "Private handler not allowed")
+ return
+
+ # Map WebSocket handler (e.g., /chat -> ws_chat)
+ handler_name = f"ws_{func_name}" if func_name else "ws_index"
+ request.path_params = parts[1:] if len(parts) > 1 else []
+ handler = getattr(self, handler_name, None)
+ if not handler:
+ await self._send_websocket_close(send, 1008, "No matching WebSocket route")
+ return
+
+ # Build function arguments
+ sig = inspect.signature(handler)
+ func_args: List[Any] = []
+ path_params_copy = request.path_params[:]
+ ws = WebSocket(receive, send)
+ func_args.append(ws) # First non-self parameter is WebSocket object
+ for param in sig.parameters.values():
+ if param.name in ("self", "ws"): # Skip self and ws parameters
+ continue
+ if param.kind == inspect.Parameter.VAR_POSITIONAL:
+ func_args.extend(path_params_copy)
+ path_params_copy = []
+ continue
+ param_value = None
+ if path_params_copy:
+ param_value = path_params_copy.pop(0)
+ elif param.name in request.query_params:
+ param_value = request.query_params[param.name][0]
+ elif param.name in request.session:
+ param_value = request.session[param.name]
+ elif param.default is not param.empty:
+ param_value = param.default
+ else:
+ await self._send_websocket_close(send, 1008, f"Missing required parameter '{param.name}'")
+ return
+ func_args.append(param_value)
+
+ # Set session ID if needed
+ session_id = cookies.get("session_id") or str(uuid.uuid4())
+ ws.session_id = session_id
+
+ # Execute handler
+ try:
+ await handler(*func_args)
+ except ConnectionClosed:
+ pass # Normal closure, no need to send another close message
+ except Exception as e:
+ print(f"WebSocket error: {e}")
+ await self._send_websocket_close(send, 1011, f"Handler error: {str(e)}")
+ return
+
+ # Save session
+ if request.session:
+ await self.session_backend.save(ws.session_id, request.session, SESSION_TIMEOUT)
+
+ finally:
+ current_request.reset(token)
+
def _parse_cookies(self, cookie_header: str) -> Dict[str, str]:
"""
Parse the Cookie header and return a dictionary of cookie names and values.
@@ -377,7 +600,7 @@ class App:
cookies[k] = v
return cookies
- async def _parse_multipart(self, reader: asyncio.StreamReader, boundary: bytes):
+ async def _parse_multipart(self, reader: asyncio.StreamReader, boundary: bytes) -> Optional[Tuple[Dict[str, List[str]], Dict[str, Any]]]:
"""
Asynchronously parses a multipart form-data request.
@@ -397,7 +620,7 @@ class App:
if not MULTIPART_INSTALLED:
print("For multipart form data support install 'multipart'.")
await self._send_response(None, 500, "500 Internal Server Error")
- return
+ return None
with PushMultipartParser(boundary) as parser:
form_data: dict = {}
@@ -507,21 +730,72 @@ class App:
await send({"type": "http.response.body", "body": b"", "more_body": False})
return
response_body = (body if isinstance(body, bytes)
- else str(body).encode("utf-8"))
+ else str(body).encode("utf-8"))
await send({
"type": "http.response.body",
"body": response_body,
"more_body": False
})
- def _redirect(self, location: str, extra_headers: list = None) -> Tuple[int, str]:
+ async def _send_websocket_response(
+ self,
+ send: Callable[[Dict[str, Any]], Awaitable[None]],
+ status_code: int,
+ body: bytes,
+ extra_headers: List[Tuple[str, str]]
+ ) -> None:
"""
- Generate an HTTP redirect response.
+ Send an HTTP response for WebSocket-related headers (e.g., cookies).
+ Args:
+ send: The ASGI send callable.
+ status_code: The HTTP status code.
+ body: The response body.
+ extra_headers: List of header tuples.
+ """
+ sanitized_headers: List[Tuple[str, str]] = []
+ for k, v in extra_headers:
+ if "\n" in k or "\r" in k or "\n" in v or "\r" in v:
+ print(f"Header injection attempt detected: {k}: {v}")
+ continue
+ sanitized_headers.append((k, v))
+ await send({
+ "type": "http.response.start",
+ "status": status_code,
+ "headers": [(k.encode("latin-1"), v.encode("latin-1")) for k, v in sanitized_headers],
+ })
+ await send({
+ "type": "http.response.body",
+ "body": body,
+ "more_body": False
+ })
+
+ async def _send_websocket_close(
+ self,
+ send: Callable[[Dict[str, Any]], Awaitable[None]],
+ code: int,
+ reason: str
+ ) -> None:
+ """
+ Send a WebSocket close message.
+
+ Args:
+ send: The ASGI send callable.
+ code: The closure code.
+ reason: The reason for closure.
+ """
+ await send({
+ "type": "websocket.close",
+ "code": code,
+ "reason": reason
+ })
+
+ def _redirect(self, location: str, extra_headers: list = None) -> Tuple[int, str, List[Tuple[str, str]]]:
+ """
+ Generate an HTTP redirect response.
Args:
location: The URL to redirect to.
extra_headers: Optional list of tuples (header_name, header_value) to include in the response.
-
Returns:
A tuple containing the HTTP status code, the HTML body, and headers list.
"""
@@ -543,7 +817,7 @@ class App:
"""
if not JINJA_INSTALLED:
print("To use the `_render_template` method install 'jinja2'.")
- return 500, "500 Internal Server Error"
+ return "500 Internal Server Error"
assert self.env is not None
template = await asyncio.to_thread(self.env.get_template, name)
return await template.render_async(**kwargs)
diff --git a/README.md b/README.md
index 86a729b..62dc8bb 100644
--- a/README.md
+++ b/README.md
@@ -141,14 +141,15 @@ class MyApp(App):
By default, MicroPie's route handlers can accept any request method, it's up to you how to handle any incoming requests! You can check the request method (and an number of other things specific to the current request state) in the handler with`self.request.method`. You can see how to handle POST JSON data at [examples/api](https://github.com/patx/micropie/tree/main/examples/api).
### Real-Time Communication with WebSockets and Socket.IO
-MicroPie is designed to be lightweight and HTTP-first, so it does not natively support WebSockets. However, because it is built directly on the ASGI specification, WebSocket support can be added via middleware or external libraries.
+MicroPie includes built-in support for WebSocket connections. WebSocket routes are defined in your App subclass using methods prefixed with `ws_`, mirroring the simplicity of MicroPie's HTTP routing. For example, a method named `ws_chat` handles WebSocket connections at `ws://<host>/chat`.
-#### Add WebSockets with Middleware
-MicroPie supports middleware that can intercept the raw ASGI scope, allowing you to handle WebSocket connections yourself. You can build full-featured WebSocket routes using a custom middleware and a subclassed App. See [examples/middleware/ws.py](https://github.com/patx/micropie/blob/main/examples/middleware/ws.py) for a working implementation of this approach, including:
+#### MicroPie’s WebSocket support allows you to:
-- Route matching with path parameters
-- Session-aware WebSocket handling
-- Chat and notification examples
+- Define WebSocket handlers with the same intuitive automatic routing as HTTP (e.g., `/chat` maps to `ws_chat` method).
+- Access query parameters, path parameters, and session data in WebSocket handlers, consistent with HTTP requests.
+- Manage WebSocket connections using the WebSocket class, which provides methods like `accept`, `receive_text`, `send_text`, and `close`.
+
+See the [websockets example](https://github.com/patx/micropie/tree/main/examples/websockets) to see how to use Websockets with MicroPie.
#### Use Socket.IO for Advanced Real-Time Features
If you want more advanced real-time features like automatic reconnection, broadcasting, or fallbacks (e.g., polling), you can integrate Socket.IO with your MicroPie app using Uvicorn as the server. See [examples/socketio](https://github.com/patx/micropie/tree/main/examples/socketio) for integration instructions and examples.
@@ -230,7 +231,7 @@ app = Root()
app.middlewares.append(MiddlewareExample())
```
-Middleware provides an easy and **reusable** way to extend the MicroPie framework. We can do things such as rate limiting, checking for max upload size in multipart requests, websockets, explicit routing, CSRF protection, and more.
+Middleware provides an easy and **reusable** way to extend the MicroPie framework. We can do things such as rate limiting, checking for max upload size in multipart requests, **explicit routing**, CSRF protection, and more.
MicroPie apps can be deployed using any ASGI server. For example, using Uvicorn if our application is saved as `app.py` and our `App` subclass is assigned to the `app` variable we can run it with:
```bash
@@ -248,8 +249,9 @@ The best way to get an idea of how MicroPie works is to see it in action! Check
- JSON Requests and Responses
- Socket.io Integration
- Async Streaming
-- Middleware including, explicit routing
+- Middleware including, explicit routing and more
- Form handling and POST requests
+- Websockets
- And more
@@ -327,7 +329,7 @@ MicroPie allows you to create pluggable middleware to hook into the request life
- `after_request(request: Request, status_code: int, response_body: Any, extra_headers: List[Tuple[str, str]]) -> None`
- Abstract method called after the request is processed but before the final response is sent to the client.
-## Request Object
+## Request Objects
### `Request` Class
@@ -345,11 +347,57 @@ Represents an HTTP request in the MicroPie framework.
- `files`: Dictionary of multipart data/streamed content.
- `headers`: Dictionary of headers.
+### `WebSocketRequest` Class
+
+Represents a WebSocket request in the MicroPie framework, inheriting from `Request`.
+
+#### Attributes
+
+- Inherits all attributes from `Request`.
+
+## WebSocket Management
+
+### `WebSocket` Class
+
+Manages WebSocket communication in the MicroPie framework.
+
+#### Methods
+
+- `__init__(receive: Callable[[], Awaitable[Dict[str, Any]]], send: Callable[[Dict[str, Any]], Awaitable[None]]) -> None`
+ - Initializes a WebSocket instance with ASGI receive and send callables.
+
+- `accept(subprotocol: Optional[str] = None, session_id: Optional[str] = None) -> None`
+ - Accepts the WebSocket connection, optionally specifying a subprotocol and session ID for setting a cookie during the handshake.
+
+- `receive_text() -> str`
+ - Receives a text message from the WebSocket. Raises `ConnectionClosed` if the connection is closed or `ValueError` for unexpected message types.
+
+- `receive_bytes() -> bytes`
+ - Receives a binary message from the WebSocket. Raises `ConnectionClosed` if the connection is closed or `ValueError` for unexpected message types.
+
+- `send_text(data: str) -> None`
+ - Sends a text message over the WebSocket. Raises `RuntimeError` if the connection is not accepted.
+
+- `send_bytes(data: bytes) -> None`
+ - Sends a binary message over the WebSocket. Raises `RuntimeError` if the connection is not accepted.
+
+- `close(code: int = 1000, reason: Optional[str] = None) -> None`
+ - Closes the WebSocket connection with the specified code and optional reason.
+
+#### Attributes
+
+- `accepted`: Boolean indicating if the WebSocket connection is accepted.
+- `session_id`: Optional string storing the session ID set during the handshake.
+
+### `ConnectionClosed` Class
+
+An exception raised when a WebSocket connection is closed.
+
## Application Base
### `App` Class
-The main ASGI application class for handling HTTP requests in MicroPie.
+The main ASGI application class for handling HTTP and WebSocket requests in MicroPie.
#### Methods
@@ -360,13 +408,16 @@ The main ASGI application class for handling HTTP requests in MicroPie.
- Retrieves the current request from the context variable.
- `__call__(scope: Dict[str, Any], receive: Callable[[], Awaitable[Dict[str, Any]]], send: Callable[[Dict[str, Any]], Awaitable[None]]) -> None`
- - ASGI callable interface for the server. Checks `scope` type.
+ - ASGI callable interface for the server. Checks `scope` type and dispatches to HTTP or WebSocket handling.
- `_asgi_app_http(scope: Dict[str, Any], receive: Callable[[], Awaitable[Dict[str, Any]]], send: Callable[[Dict[str, Any]], Awaitable[None]]) -> None`
- ASGI application entry point for handling HTTP requests.
+- `_asgi_app_websocket(scope: Dict[str, Any], receive: Callable[[], Awaitable[Dict[str, Any]]], send: Callable[[Dict[str, Any]], Awaitable[None]]) -> None`
+ - ASGI application entry point for handling WebSocket requests. Routes to methods prefixed with `ws_` (e.g., `ws_chat` for `/chat`).
+
- `request(self) -> Request`
- - Accessor for the current request object. - Returns the current request from the context variable.
+ - Accessor for the current request object. Returns the current request from the context variable.
- `_parse_cookies(cookie_header: str) -> Dict[str, str]`
- Parses the Cookie header and returns a dictionary of cookie names and values.
@@ -382,6 +433,9 @@ The main ASGI application class for handling HTTP requests in MicroPie.
- `_send_response(send: Callable[[Dict[str, Any]], Awaitable[None]], status_code: int, body: Any, extra_headers: Optional[List[Tuple[str, str]]] = None) -> None`
- Sends an HTTP response using the ASGI send callable.
+- `_send_websocket_close(send: Callable[[Dict[str, Any]], Awaitable[None]], code: int, reason: str) -> None`
+ - Sends a WebSocket close message with the specified code and reason.
+
- `_redirect(location: str) -> Tuple[int, str]`
- Generates an HTTP redirect response.
@@ -389,7 +443,7 @@ The main ASGI application class for handling HTTP requests in MicroPie.
- Renders a template asynchronously using Jinja2.
- *Requires*: `jinja2`
-The `App` class is the main entry point for creating MicroPie applications. It implements the ASGI interface and handles HTTP requests.
+The `App` class is the main entry point for creating MicroPie applications. It implements the ASGI interface and handles HTTP and WebSocket requests.
## Response Formats
@@ -410,6 +464,6 @@ MicroPie provides built-in error handling for common HTTP status codes:
Custom error handling can be implemented through middleware.
-----
+---
-© 2025 Harrison Erd
+© 2025 Harrison Erd
\ No newline at end of file
diff --git a/examples/middleware/ws.py b/examples/middleware/ws.py
deleted file mode 100644
index 50f0944..0000000
--- a/examples/middleware/ws.py
+++ /dev/null
@@ -1,207 +0,0 @@
-"""
-Example of a WebSocket middleware for MicroPie that handles WebSocket connections.
-
-This middleware enables WebSocket support by intercepting WebSocket requests,
-managing the connection lifecycle, and routing messages to appropriate handlers.
-It includes a specialized WebSocketRequest class to handle WebSocket scopes.
-
-For comprehensive WebSocket examples, refer to the MicroPie documentation at
-https://patx.github.io/micropie
-"""
-
-import re
-import json
-import asyncio
-from typing import Dict, List, Optional, Tuple, Any, Callable, Awaitable
-from MicroPie import App, HttpMiddleware, Request, current_request
-from urllib.parse import parse_qs
-
-class WebSocketRequest:
- """Represents a WebSocket request in the MicroPie framework."""
- def __init__(self, scope: Dict[str, Any]) -> None:
- """
- Initialize a new WebSocketRequest instance.
-
- Args:
- scope: The ASGI scope dictionary for the WebSocket request.
- """
- self.scope: Dict[str, Any] = scope
- self.path_params: List[str] = []
- self.query_params: Dict[str, List[str]] = {}
- self.session: Dict[str, Any] = {}
- self.headers: Dict[str, str] = {
- k.decode("utf-8", errors="replace").lower(): v.decode("utf-8", errors="replace")
- for k, v in scope.get("headers", [])
- }
- # Parse query parameters
- self.query_params = parse_qs(scope.get("query_string", b"").decode("utf-8", "ignore"))
-
-class WebSocketMiddleware(HttpMiddleware):
- def __init__(self):
- # Map WebSocket paths to handler method names
- self.ws_routes: Dict[str, Tuple[str, str]] = {}
-
- def add_ws_route(self, path: str, handler_name: str) -> None:
- """
- Register a WebSocket route with its handler method name.
-
- Args:
- path: The WebSocket route pattern (e.g., "/ws/chat/{room}")
- handler_name: The handler method name (e.g., "_handle_chat")
- """
- pattern = re.sub(r"{([^}]+)}", r"([^/]+)", path)
- pattern = f"^{pattern}$"
- self.ws_routes[path] = (pattern, handler_name)
-
- async def before_request(self, request: Request) -> Optional[Dict]:
- """
- Skip WebSocket requests in the HTTP middleware pipeline.
-
- Args:
- request: The MicroPie Request object
-
- Returns:
- None to let MicroPie handle requests
- """
- if request.scope["type"] == "websocket":
- return None # WebSocket requests are handled in _handle_websocket
- return None
-
- async def after_request(
- self,
- request: Request,
- status_code: int,
- response_body: Any,
- extra_headers: List[Tuple[str, str]]
- ) -> Optional[Dict]:
- return None
-
-class WebSocketApp(App):
- def __init__(self):
- super().__init__()
- self.ws_middleware = WebSocketMiddleware()
- self.middlewares.append(self.ws_middleware)
-
- # Register WebSocket routes
- self.ws_middleware.add_ws_route("/ws/chat/{room}", "_handle_chat")
- self.ws_middleware.add_ws_route("/ws/notifications", "_handle_notifications")
-
- async def __call__(
- self,
- scope: Dict[str, Any],
- receive: Callable[[], Awaitable[Dict[str, Any]]],
- send: Callable[[Dict[str, Any]], Awaitable[None]]
- ) -> None:
- """
- Override the App's ASGI callable to handle WebSocket connections.
- """
- if scope["type"] == "websocket":
- await self._handle_websocket(scope, receive, send)
- else:
- await super().__call__(scope, receive, send)
-
- async def _handle_websocket(
- self,
- scope: Dict[str, Any],
- receive: Callable[[], Awaitable[Dict[str, Any]]],
- send: Callable[[Dict[str, Any]], Awaitable[None]]
- ) -> None:
- """
- Handle WebSocket connections by routing to the appropriate handler.
-
- Args:
- scope: The ASGI scope dictionary
- receive: The callable to receive ASGI events
- send: The callable to send ASGI events
- """
- request = WebSocketRequest(scope)
- token = current_request.set(request)
-
- try:
- # Perform WebSocket route matching
- path = scope["path"]
- handler = None
- path_params = []
- for route_path, (pattern, handler_name) in self.ws_middleware.ws_routes.items():
- match = re.match(pattern, path)
- if match:
- path_params = [str(param) for param in match.groups()]
- handler = getattr(self, handler_name, None)
- break
-
- if not handler:
- await send({
- "type": "websocket.close",
- "code": 1008, # Policy violation
- "reason": "No matching WebSocket route"
- })
- return
-
- # Accept the WebSocket connection
- await send({
- "type": "websocket.accept",
- "subprotocol": None,
- })
-
- # Execute the WebSocket handler with path parameters
- try:
- await handler(request, receive, send, *path_params)
- except Exception as e:
- print(f"WebSocket error: {e}")
- await send({
- "type": "websocket.close",
- "code": 1011, # Internal error
- "reason": f"Handler error: {str(e)}"
- })
-
- finally:
- current_request.reset(token)
-
- async def _handle_chat(self, request: WebSocketRequest, receive: Callable, send: Callable, room: str):
- """
- Handle WebSocket connections for a chat room.
-
- Args:
- request: The WebSocketRequest object
- receive: The ASGI receive callable
- send: The ASGI send callable
- room: The chat room identifier from the path
- """
- while True:
- message = await receive()
- if message["type"] == "websocket.disconnect":
- break
- if message["type"] == "websocket.receive":
- data = message.get("text") or message.get("bytes", b"").decode("utf-8", "ignore")
- response = json.dumps({"room": room, "message": data, "type": "chat.message"})
- await send({
- "type": "websocket.send",
- "text": response
- })
-
- async def _handle_notifications(self, request: WebSocketRequest, receive: Callable, send: Callable):
- """
- Handle WebSocket connections for sending notifications.
-
- Args:
- request: The WebSocketRequest object
- receive: The ASGI receive callable
- send: The ASGI send callable
- """
- try:
- count = 0
- while True:
- await asyncio.sleep(5) # Send notification every 5 seconds
- response = json.dumps({"notification": f"Update #{count}", "type": "notification"})
- await send({
- "type": "websocket.send",
- "text": response
- })
- count += 1
- except asyncio.CancelledError:
- await send({
- "type": "websocket.close",
- "code": 1000, # Normal closure
- })
-
-app = WebSocketApp()
diff --git a/examples/websockets/app.py b/examples/websockets/app.py
new file mode 100644
index 0000000..5a6a94c
--- /dev/null
+++ b/examples/websockets/app.py
@@ -0,0 +1,21 @@
+from MicroPie import App, ConnectionClosed
+
+class MyApp(App):
+ async def chat(self):
+ """HTTP handler for GET /chat"""
+ return "Welcome to MicroPie, this is an HTTP GET route!"
+
+ async def ws_chat(self, ws, room=None):
+ """WebSocket handler for ws://localhost:8000/chat"""
+ await ws.accept()
+ user = self.request.query_params.get("user", ["anonymous"])[0]
+ self.request.session["last_room"] = room or "default"
+ while True:
+ try:
+ message = await ws.receive_text()
+ response = f"{user} in {room or 'default'}: {message}"
+ await ws.send_text(response)
+ except ConnectionClosed:
+ break
+
+app = MyApp()
diff --git a/examples/websockets/test.html b/examples/websockets/test.html
new file mode 100644
index 0000000..46e8770
--- /dev/null
+++ b/examples/websockets/test.html
@@ -0,0 +1,25 @@
+<!DOCTYPE html>
+<html>
+<head>
+ <title>WebSocket Test</title>
+</head>
+<body>
+ <input id="message" type="text">
+ <button onclick="sendMessage()">Send</button>
+ <div id="output"></div>
+ <script>
+ const ws = new WebSocket("ws://localhost:8000/chat?user=Alice");
+ ws.onmessage = function(event) {
+ document.getElementById("output").innerText += event.data + "\n";
+ };
+ ws.onclose = function(event) {
+ document.getElementById("output").innerText += `Closed: ${event.code} ${event.reason}\n`;
+ };
+ function sendMessage() {
+ const msg = document.getElementById("message").value;
+ ws.send(msg);
+ document.getElementById("message").value = "";
+ }
+ </script>
+</body>
+</html>
diff --git a/pyproject.toml b/pyproject.toml
index 263d012..88fa6b0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "flit_core.buildapi"
[project]
name = "MicroPie"
-version = "0.12.2"
+version = "0.13"
description = "An ultra micro ASGI web framework"
keywords = ["micropie", "asgi", "microframework", "http"]
readme = "docs/README.md"