src/liquidcode/websocket/websocket_request.py - develop-0.2.0 - liquid-g/liquid-code

src/liquidcode/websocket/websocket_request.py

70 строк2 KB

User

ci: настроен Black и flake8 для автоматического форматирования кода

04 июл 2026, 08:14

04 июл 2026, 08:144778ed6

"""
WebSocket-реализация абстрактного Request.
"""

from typing import Any, Dict, Optional
from urllib.parse import parse_qs, urlparse

from ..contracts import Request


class WebSocketRequest(Request):
    """
    Реализация Request для WebSocket-протокола.
    """

    def __init__(
        self,
        path: str,
        message: str,
        websocket,
        headers: Optional[Dict[str, str]] = None,
        query_params: Optional[Dict[str, str]] = None,
    ):
        self._path = path
        self._message = message
        self._websocket = websocket
        self._headers = headers or {}
        self._route_params: Dict[str, str] = {}
        self._attrs: Dict[str, Any] = {}

        if query_params is not None:
            self._query_params = query_params
        else:
            self._query_params = self._parse_query_params(path)

    def _parse_query_params(self, path: str) -> Dict[str, str]:
        parsed = urlparse(path)
        query = parse_qs(parsed.query)
        return {k: v[-1] if v else "" for k, v in query.items()}

    def get_method(self) -> str:
        return "WEBSOCKET"

    def get_path(self) -> str:
        return self._path

    def get_headers(self) -> Dict[str, str]:
        return self._headers

    def get_body(self) -> bytes:
        return self._message.encode("utf-8")

    def get_query_params(self) -> Dict[str, str]:
        return self._query_params

    def get_route_params(self) -> Dict[str, str]:
        return self._route_params

    def set_route_params(self, params: Dict[str, str]) -> None:
        self._route_params = params

    def get_attr(self, key: str, default: Any = None) -> Any:
        return self._attrs.get(key, default)

    def set_attr(self, key: str, value: Any) -> None:
        self._attrs[key] = value

    def get_websocket(self):
        """Возвращает объект WebSocket-соединения."""
        return self._websocket