# client_wss.py
# pylint: disable=line-too-long,broad-exception-raised,broad-exception-caught,import-outside-toplevel,protected-access
# Standard library imports
from http import HTTPStatus
from pathlib import Path
from typing import Optional, Tuple, Union
from urllib import parse
# Third-party imports
import requests
# Local imports
from .client_http import AppMeshClient
from .exceptions import AppMeshRequestError
from .token_provider import TokenProvider
from .wss_transport import WSSTransport
from .transport_mixin import TransportClientMixin
[docs]
class AppMeshClientWSS(TransportClientMixin, AppMeshClient):
"""
App Mesh client over WebSocket Secure (WSS).
Same API as ``AppMeshClient`` but overrides file up/download to use a WSS
side channel for faster large-file transfers, and supports event subscription.
Methods:
# Overridden for the WSS transport
- download_file()
- upload_file()
- add_app() # subscribe atomically when the app starts
- wait_for_async_run() # subscribe-based output streaming
- close()
# Event Subscription (TCP/WSS only)
- subscribe()
- unsubscribe()
Inherits all other methods from AppMeshClient.
Example:
>>> from appmesh import AppMeshClientWSS
>>> client = AppMeshClientWSS(bearer_token="access-token")
>>> client.download_file("/tmp/os-release", "os-release")
"""
# WSS-optimized chunk size
_WSS_BLOCK_SIZE = 64 * 1024
_HTTP_USER_AGENT_TRANSPORT = "appmesh/python/wss"
def __init__(
self,
wss_address: Tuple[str, int] = ("127.0.0.1", 6058),
ssl_verify: Union[bool, str, None] = None,
ssl_client_cert: Optional[Union[str, Tuple[str, str]]] = None,
*,
bearer_token: Optional[str] = None,
token_provider: Optional[TokenProvider] = None,
):
"""Construct a WSS transport client that reuses the standard App Mesh client API.
Args:
wss_address: Server address as (host, port) tuple, defaults to ("127.0.0.1", 6058).
ssl_verify: SSL certificate verification behavior. Can be None, True, False, or a path to CA bundle.
- None (default): Auto — use the App Mesh CA bundle if installed, otherwise system CAs
- True: Use system CA certificates (e.g., /etc/ssl/certs/ on Linux)
- False: Disable verification (insecure, must be requested explicitly)
- str: Path to custom CA bundle or directory (must exist)
ssl_client_cert: SSL client certificate:
- str: Path to single PEM with cert+key
- tuple: (cert_path, key_path)
bearer_token: Caller-owned access token.
token_provider: Provider that supplies and refreshes access tokens.
Note:
WSS connections require an explicit full-chain CA specification for certificate validation,
unlike HTTP, which can retrieve intermediate certificates automatically.
"""
ssl_verify = AppMeshClient._resolve_ssl_verify(ssl_verify)
self.wss_transport = WSSTransport(address=wss_address, ssl_verify=ssl_verify, ssl_client_cert=ssl_client_cert)
self._transport_client_addr = "wss-client"
self._transport_name = "WebSocket"
# http and websocket share same address
host, port = wss_address
super().__init__(
base_url=f"https://{host}:{port}",
ssl_verify=ssl_verify,
ssl_client_cert=ssl_client_cert,
bearer_token=bearer_token,
token_provider=token_provider,
)
@property
def _transport(self):
"""Return the WSS transport instance."""
return self.wss_transport
[docs]
def close(self) -> None:
"""Close the connection and release resources."""
if self._demuxer:
self._demuxer.stop()
if hasattr(self, "wss_transport") and self.wss_transport:
self.wss_transport.close()
self.wss_transport = None
if self._demuxer:
self._demuxer.join()
self._demuxer = None
return super().close()
def __del__(self):
"""Ensure resources are properly released when the object is garbage collected."""
try:
self.close()
except Exception:
pass # Never raise in __del__
[docs]
def download_file(self, remote_file: str, local_file: Optional[str] = None, preserve_permissions: bool = False) -> None:
"""Copy a remote file to local through the WSS control channel plus HTTPS data channel.
Args:
remote_file: Remote file path.
local_file: Local destination path; defaults to the remote file's basename.
preserve_permissions: Apply remote file permissions/ownership locally on a best-effort basis.
"""
if not local_file:
local_file = Path(remote_file).name
header = {AppMeshClient._HTTP_HEADER_KEY_X_FILE_PATH: parse.quote(remote_file)}
# Control-channel precheck (permission/existence); raises on failure.
self._request_http(AppMeshClient._Method.GET, path="/appmesh/file/download", header=header)
token = self._get_bearer_token()
if not token:
raise ValueError("File transfer requires a bearer token")
# Use requests to GET file
local_path = Path(local_file)
header = {
AppMeshClient._HTTP_HEADER_KEY_X_FILE_PATH: parse.quote(remote_file),
AppMeshClient._HTTP_HEADER_KEY_AUTH: f"Bearer {token}",
}
path = "/appmesh/file/download/ws"
rest_url = parse.urljoin(self.base_url, path)
r = self.session.get(url=rest_url, stream=True, timeout=self.request_timeout, headers=header, cert=self.ssl_client_cert, verify=self.ssl_verify)
if r.status_code == HTTPStatus.OK:
# Write file in chunks
with local_path.open("wb") as fp:
for chunk in r.iter_content(chunk_size=self._WSS_BLOCK_SIZE):
if chunk:
fp.write(chunk)
# Apply file attributes if requested
if preserve_permissions:
AppMeshClient._apply_file_attributes(local_path, r.headers)
else:
try:
r.raise_for_status()
except requests.exceptions.HTTPError as e:
raise AppMeshRequestError(f"File download failed: {e}") from e
[docs]
def upload_file(self, local_file: str, remote_file: Optional[str] = None, preserve_permissions: bool = False) -> None:
"""Upload a local file through the WSS control channel plus HTTPS data channel.
Args:
local_file: Local file path.
remote_file: Remote destination path; defaults to the local file's basename.
preserve_permissions: Send local file permissions/ownership metadata when available.
"""
if not remote_file:
remote_file = Path(local_file).name
header = {AppMeshClient._HTTP_HEADER_KEY_X_FILE_PATH: parse.quote(remote_file)}
# Control-channel precheck (permission/existence); raises on failure.
self._request_http(AppMeshClient._Method.POST, path="/appmesh/file/upload", header=header)
token = self._get_bearer_token()
if not token:
raise ValueError("File transfer requires a bearer token")
local_path = Path(local_file)
if not local_path.exists():
raise FileNotFoundError(f"Local file not found: {local_file}")
# Upload file with http
path = "/appmesh/file/upload/ws"
header = {
AppMeshClient._HTTP_HEADER_KEY_AUTH: f"Bearer {token}",
AppMeshClient._HTTP_HEADER_KEY_X_FILE_PATH: parse.quote(remote_file),
}
if preserve_permissions:
header.update(AppMeshClient._get_file_attributes(local_path))
rest_url = parse.urljoin(self.base_url, path)
with local_path.open("rb") as fp:
r = self.session.post(url=rest_url, stream=True, data=fp, timeout=self.request_timeout, headers=header, cert=self.ssl_client_cert, verify=self.ssl_verify)
try:
r.raise_for_status()
except requests.exceptions.HTTPError as e:
raise AppMeshRequestError(f"File upload failed: {e}") from e