qpanda3_runtime.access_service.network#

Network communication module for quantum computing runtime service.

This module provides HTTP and TCP client implementations for communicating with quantum computing backend services. It supports both synchronous and asynchronous communication patterns.

Modules#

http_clientHTTPClient

Asynchronous HTTP client for REST API communication with quantum backend.

query_task_tcpTCPQueryTask

TCP-based client for querying quantum task execution results.

Notes

  • The HTTP client supports SSL/TLS with certificate verification disabled for development environments.

  • The TCP client implements custom binary protocol for efficient data transfer.

  • Both clients are designed for high-throughput quantum task management.

Examples

>>> from .http_client import HTTPClient
>>> client = HTTPClient(base_url='http://localhost:8080')
>>> result = client.get_request('/api/status')

>>> from .query_task_tcp import TCPQueryTask
>>> result = TCPQueryTask.query_task_tcp('localhost', 8080, 'task-123')

Submodules#

Classes#

HTTPClient

Asynchronous HTTP client for quantum computing backend communication.

TCPQueryTask

TCP client for querying quantum task execution results.

Package Contents#

class qpanda3_runtime.access_service.network.HTTPClient(base_url='http://localhost:8080', outbound_raw_requests_logfile: str | None = None)#

Asynchronous HTTP client for quantum computing backend communication.

Provides synchronous and asynchronous methods for making HTTP requests to quantum computing backend services. Supports request logging and SSL/TLS configuration.

Parameters:

  • base_url (str, optional) – Base URL for all HTTP requests (default: ‘http://localhost:8080’).

  • outbound_raw_requests_logfile (Union[str, None], optional) – Path to log file for recording outbound request data. If provided, all request payloads will be logged to this file.

Variables:

  • _base_url (str) – Base URL for HTTP requests.

  • _outbound_raw_requests_logfile (Union[str, None]) – Path to request logging file.

Notes

  • SSL certificate verification is disabled for development environments.

  • All requests use JSON format for both request and response.

  • Timeout handling is managed by aiohttp default settings.

Examples

>>> client = HTTPClient(base_url='http://quantum-backend:8080')
>>> result = client.get_request('/api/status')
>>> async_result = await client.post_request_async('/api/task', {'task_id': '123'})
get_request(endpoint)#

Synchronous GET request.

Parameters:

endpoint (str) – API endpoint path.

Returns:

Parsed JSON response.

Return type:

dict

See also

_make_request

Internal request implementation.

post_request(endpoint, data)#

Synchronous POST request.

Parameters:

  • endpoint (str) – API endpoint path.

  • data (dict) – JSON-serializable request data.

Returns:

Parsed JSON response.

Return type:

dict

async post_request_async(endpoint, data, extra_headers: dict = None)#

Asynchronous POST request.

Parameters:

  • endpoint (str) – API endpoint path.

  • data (dict) – JSON-serializable request data.

  • extra_headers (dict, optional) – Additional HTTP headers to include in the request.

Returns:

Parsed JSON response.

Return type:

dict

Notes

Use this method in async contexts to avoid blocking.

put_request(endpoint, data)#

Synchronous PUT request.

Parameters:

  • endpoint (str) – API endpoint path.

  • data (dict) – JSON-serializable request data.

Returns:

Parsed JSON response.

Return type:

dict

delete_request(endpoint)#

Synchronous DELETE request.

Parameters:

endpoint (str) – API endpoint path.

Returns:

Parsed JSON response.

Return type:

dict

class qpanda3_runtime.access_service.network.TCPQueryTask#

TCP client for querying quantum task execution results.

Implements custom binary protocol over TCP for efficient querying of quantum computing task results. Includes heartbeat mechanism to keep connections alive during long-running queries.

Notes

  • Uses port+1 for TCP queries (e.g., if HTTP port is 8080, TCP port is 8081).

  • Implements simple timeout mechanism based on non-target message count.

  • Heartbeat daemon runs in background thread to prevent connection timeout.

Examples

>>> result = TCPQueryTask.query_task_tcp('localhost', 8080, 'task-123')
>>> print(result)  # JSON string with task results
static drain_msg(client_socket, res: memoryview, msg_len: int, received_byte_total)#

Drain complete message from socket into buffer.

Parameters:

  • client_socket (socket.socket) – Connected TCP socket.

  • res (memoryview) – Memory view of target buffer.

  • msg_len (int) – Total expected message length.

  • received_byte_total (int) – Bytes already received.

Returns:

Total bytes received after draining.

Return type:

int

static start_heartbeat_daemon_no_lock(shared_socket, heartmsg: bytes, interval)#

Start heartbeat daemon thread for connection keep-alive.

Parameters:

  • shared_socket (socket.socket) – Socket to send heartbeat messages on.

  • heartmsg (bytes) – Heartbeat message bytes.

  • interval (int) – Heartbeat interval in seconds.

Notes

  • Daemon thread automatically stops when socket connection breaks.

  • No synchronization locks used (assumes single-threaded socket access).

static query_task_tcp(host: str, port: int, task_id: str, timeout=50)#

Query quantum task results via TCP protocol.

Establishes TCP connection, sends query message, waits for result, and returns decoded response. Includes error handling and timeout management.

Parameters:

  • host (str) – Server hostname or IP address.

  • port (int) – Base HTTP port (TCP port will be port+1).

  • task_id (str) – Unique task identifier to query.

  • timeout (int, optional) – Socket timeout in seconds (default: 50).

Returns:

JSON string containing task results.

Return type:

str

Raises:

  • ConnectionRefusedError – If server is not running or not listening.

  • socket.gaierror – If hostname resolution fails.

  • RuntimeError – If timeout occurs or protocol error detected.

Notes

  • TCP connection uses port+1 (e.g., HTTP port 8080 → TCP port 8081).

  • Socket options optimized for low latency (TCP_NODELAY, TCP_QUICKACK).

  • Timeout based on count of non-result messages received.

  • Heartbeat daemon prevents connection idle timeout.

Examples

>>> result = TCPQueryTask.query_task_tcp('localhost', 8080, 'task-123')
>>> import json
>>> data = json.loads(result)