Source code for appmesh.client_wss

# 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