Skip to content

Transport

Overview

The transport layer abstracts HTTP communication from the session logic. The session builds runCommandJSON payloads and delegates HTTP delivery to a transport implementation. This separation enables testing the entire command pipeline without an MQ server by injecting a mock transport.

MQRESTTransport

The transport protocol defines a single method for posting JSON payloads:

Bases: Protocol

Protocol for MQ REST transport implementations.

Implement this protocol to provide a custom HTTP transport (e.g. for testing or for an alternative HTTP client library). The default implementation is :class:RequestsTransport.

Source code in src/pymqrest/session.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
class MQRESTTransport(Protocol):
    """Protocol for MQ REST transport implementations.

    Implement this protocol to provide a custom HTTP transport
    (e.g. for testing or for an alternative HTTP client library).
    The default implementation is :class:`RequestsTransport`.
    """

    def post_json(
        self,
        url: str,
        payload: Mapping[str, object],
        *,
        headers: Mapping[str, str],
        timeout_seconds: float | None,
        verify_tls: bool,
    ) -> TransportResponse:
        """Send a JSON payload via HTTP POST and return the response.

        Args:
            url: The fully-qualified URL to POST to.
            payload: The JSON-serialisable request body.
            headers: HTTP headers to include in the request.
            timeout_seconds: Request timeout in seconds, or ``None``
                for no timeout.
            verify_tls: Whether to verify the server's TLS certificate.

        Returns:
            A :class:`TransportResponse` with the status code, body
            text, and response headers.

        Raises:
            MQRESTTransportError: If the request cannot be completed.

        """

post_json(url, payload, *, headers, timeout_seconds, verify_tls)

Send a JSON payload via HTTP POST and return the response.

Parameters:

Name Type Description Default
url str

The fully-qualified URL to POST to.

 required
payload Mapping[str, object]

The JSON-serialisable request body.

 required
headers Mapping[str, str]

HTTP headers to include in the request.

 required
timeout_seconds float | None

Request timeout in seconds, or None for no timeout.

 required
verify_tls bool

Whether to verify the server's TLS certificate.

 required

Returns:

Name Type Description
A TransportResponse

class:TransportResponse with the status code, body
TransportResponse

text, and response headers.

Raises:

Type Description
MQRESTTransportError

If the request cannot be completed.
Source code in src/pymqrest/session.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def post_json(
    self,
    url: str,
    payload: Mapping[str, object],
    *,
    headers: Mapping[str, str],
    timeout_seconds: float | None,
    verify_tls: bool,
) -> TransportResponse:
    """Send a JSON payload via HTTP POST and return the response.

    Args:
        url: The fully-qualified URL to POST to.
        payload: The JSON-serialisable request body.
        headers: HTTP headers to include in the request.
        timeout_seconds: Request timeout in seconds, or ``None``
            for no timeout.
        verify_tls: Whether to verify the server's TLS certificate.

    Returns:
        A :class:`TransportResponse` with the status code, body
        text, and response headers.

    Raises:
        MQRESTTransportError: If the request cannot be completed.

    """

TransportResponse

An immutable result containing the HTTP response data:

Container for the raw HTTP response returned by a transport.

Attributes:

Name Type Description
status_code int

The HTTP status code (e.g. 200, 401).
text str

The response body as text.
headers Mapping[str, str]

The response headers as a string-to-string mapping.
Source code in src/pymqrest/session.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@dataclass(frozen=True)
class TransportResponse:
    """Container for the raw HTTP response returned by a transport.

    Attributes:
        status_code: The HTTP status code (e.g. ``200``, ``401``).
        text: The response body as text.
        headers: The response headers as a string-to-string mapping.

    """

    status_code: int
    text: str
    headers: Mapping[str, str]

status_code instance-attribute

text instance-attribute

headers instance-attribute

__init__(status_code, text, headers)

RequestsTransport

The default transport implementation using the requests library:

Default :class:MQRESTTransport implementation using requests.

Wraps a :class:requests.Session to handle JSON POST requests to the MQ REST API. Connection-level errors are translated into :class:~pymqrest.exceptions.MQRESTTransportError.

Source code in src/pymqrest/session.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class RequestsTransport:
    """Default :class:`MQRESTTransport` implementation using ``requests``.

    Wraps a :class:`requests.Session` to handle JSON POST requests
    to the MQ REST API. Connection-level errors are translated into
    :class:`~pymqrest.exceptions.MQRESTTransportError`.
    """

    def __init__(
        self,
        session: requests.Session | None = None,
        *,
        client_cert: tuple[str, str] | str | None = None,
    ) -> None:
        """Initialize the transport.

        Args:
            session: An existing :class:`requests.Session` to reuse,
                or ``None`` to create a new one.
            client_cert: Client certificate for mutual TLS. Either a
                path to a combined cert/key PEM file, or a
                ``(cert_path, key_path)`` tuple.

        """
        self._session = session or requests.Session()
        if client_cert is not None:
            self._session.cert = client_cert

    def post_json(
        self,
        url: str,
        payload: Mapping[str, object],
        *,
        headers: Mapping[str, str],
        timeout_seconds: float | None,
        verify_tls: bool,
    ) -> TransportResponse:
        """Send a JSON payload via HTTP POST and return the response.

        Args:
            url: The fully-qualified URL to POST to.
            payload: The JSON-serialisable request body.
            headers: HTTP headers to include in the request.
            timeout_seconds: Request timeout in seconds, or ``None``
                for no timeout.
            verify_tls: Whether to verify the server's TLS certificate.

        Returns:
            A :class:`TransportResponse` with the status code, body
            text, and response headers.

        Raises:
            MQRESTTransportError: If the underlying ``requests`` call
                raises a :class:`~requests.RequestException`.

        """
        try:
            response = self._session.post(
                url,
                json=dict(payload),
                headers=dict(headers),
                timeout=timeout_seconds,
                verify=verify_tls,
            )
        except RequestException as error:
            raise MQRESTTransportError(ERROR_TRANSPORT_FAILURE, url=url) from error
        return TransportResponse(
            status_code=response.status_code,
            text=response.text,
            headers=dict(response.headers.items()),
        )

_session = session or requests.Session() instance-attribute

__init__(session=None, *, client_cert=None)

Initialize the transport.

Parameters:

Name Type Description Default
session Session | None

An existing :class:requests.Session to reuse, or None to create a new one.

 None
client_cert tuple[str, str] | str | None

Client certificate for mutual TLS. Either a path to a combined cert/key PEM file, or a (cert_path, key_path) tuple.

 None
Source code in src/pymqrest/session.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def __init__(
    self,
    session: requests.Session | None = None,
    *,
    client_cert: tuple[str, str] | str | None = None,
) -> None:
    """Initialize the transport.

    Args:
        session: An existing :class:`requests.Session` to reuse,
            or ``None`` to create a new one.
        client_cert: Client certificate for mutual TLS. Either a
            path to a combined cert/key PEM file, or a
            ``(cert_path, key_path)`` tuple.

    """
    self._session = session or requests.Session()
    if client_cert is not None:
        self._session.cert = client_cert

post_json(url, payload, *, headers, timeout_seconds, verify_tls)

Send a JSON payload via HTTP POST and return the response.

Parameters:

Name Type Description Default
url str

The fully-qualified URL to POST to.

 required
payload Mapping[str, object]

The JSON-serialisable request body.

 required
headers Mapping[str, str]

HTTP headers to include in the request.

 required
timeout_seconds float | None

Request timeout in seconds, or None for no timeout.

 required
verify_tls bool

Whether to verify the server's TLS certificate.

 required

Returns:

Name Type Description
A TransportResponse

class:TransportResponse with the status code, body
TransportResponse

text, and response headers.

Raises:

Type Description
MQRESTTransportError

If the underlying requests call raises a :class:~requests.RequestException.
Source code in src/pymqrest/session.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def post_json(
    self,
    url: str,
    payload: Mapping[str, object],
    *,
    headers: Mapping[str, str],
    timeout_seconds: float | None,
    verify_tls: bool,
) -> TransportResponse:
    """Send a JSON payload via HTTP POST and return the response.

    Args:
        url: The fully-qualified URL to POST to.
        payload: The JSON-serialisable request body.
        headers: HTTP headers to include in the request.
        timeout_seconds: Request timeout in seconds, or ``None``
            for no timeout.
        verify_tls: Whether to verify the server's TLS certificate.

    Returns:
        A :class:`TransportResponse` with the status code, body
        text, and response headers.

    Raises:
        MQRESTTransportError: If the underlying ``requests`` call
            raises a :class:`~requests.RequestException`.

    """
    try:
        response = self._session.post(
            url,
            json=dict(payload),
            headers=dict(headers),
            timeout=timeout_seconds,
            verify=verify_tls,
        )
    except RequestException as error:
        raise MQRESTTransportError(ERROR_TRANSPORT_FAILURE, url=url) from error
    return TransportResponse(
        status_code=response.status_code,
        text=response.text,
        headers=dict(response.headers.items()),
    )

Custom transport

Implement the MQRESTTransport protocol to provide custom HTTP behavior or for testing. Because the protocol has a single method, a mock works naturally:

from unittest.mock import MagicMock
from pymqrest.session import MQRESTTransport, TransportResponse

mock_transport = MagicMock(spec=MQRESTTransport)
mock_transport.post_json.return_value = TransportResponse(
    status_code=200,
    text='{"commandResponse": []}',
    headers={},
)

session = MQRESTSession(
    rest_base_url="https://localhost:9443/ibmmq/rest/v2",
    qmgr_name="QM1",
    credentials=LTPAAuth("admin", "passw0rd"),
    transport=mock_transport,
)

This pattern is used extensively in the library's own test suite to verify command payload construction, response parsing, and error handling without network access.