Session

Overview

The MQRESTSession class is the main entry point for interacting with IBM MQ via the REST API. A session encapsulates connection details, authentication, attribute mapping configuration, and diagnostic state. It inherits MQSC command methods from MQRESTCommandMixin (see commands) and idempotent ensure methods from MQRESTEnsureMixin (see ensure).

Creating a session

from pymqrest import MQRESTSession
from pymqrest.auth import LTPAAuth

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

The constructor validates all required fields and constructs the transport, mapping data, and authentication state at creation time. Errors in configuration (e.g. invalid mapping overrides) are caught immediately.

Constructor parameters

Parameter	Type	Description
rest_base_url	Required	Base URL of the MQ REST API (e.g. https://host:9443/ibmmq/rest/v2)
qmgr_name	Required	Target queue manager name
credentials	Required	Authentication credentials (LTPAAuth, BasicAuth, or CertificateAuth)
gateway_qmgr	Optional	Gateway queue manager for remote routing
map_attributes	Optional	Enable/disable attribute mapping (default: True)
mapping_strict	Optional	Strict or lenient mapping mode (default: True)
mapping_overrides	Optional	Custom mapping overrides (sparse merge)
verify_tls	Optional	Verify server TLS certificates (default: True)
timeout	Optional	Default request timeout in seconds
csrf_token	Optional	Custom CSRF token value
transport	Optional	Custom transport implementation

Minimal example

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

Full example

session = MQRESTSession(
    rest_base_url="https://mq-server.example.com:9443/ibmmq/rest/v2",
    qmgr_name="QM2",
    credentials=LTPAAuth("mqadmin", "mqadmin"),
    gateway_qmgr="QM1",
    map_attributes=True,
    mapping_strict=False,
    mapping_overrides=overrides,
    verify_tls=True,
    timeout=30,
)

Command methods

The session provides ~144 command methods, one for each MQSC verb + qualifier combination. See Commands for the full list.

# DISPLAY commands return a list of dicts
queues = session.display_queue("APP.*")

# Queue manager singletons return a single dict or None
qmgr = session.display_qmgr()

# Non-DISPLAY commands return None (raise on error)
session.define_qlocal("MY.QUEUE", request_parameters={"max_queue_depth": 50000})
session.delete_queue("MY.QUEUE")

Ensure methods

The session provides 16 ensure methods for declarative object management. Each method implements an idempotent upsert: DEFINE if the object does not exist, ALTER only the attributes that differ, or no-op if already correct.

result = session.ensure_qlocal("MY.QUEUE",
    request_parameters={"max_queue_depth": 50000})
# result.action is EnsureAction.CREATED, UPDATED, or UNCHANGED

See Ensure for detailed usage and the full list of available ensure methods.

Diagnostic state

The session retains the most recent request and response for inspection. This is useful for debugging command failures or understanding what the library sent to the MQ REST API:

session.display_queue("MY.QUEUE")

session.last_command_payload    # the JSON sent to MQ (dict)
session.last_response_payload   # the parsed JSON response (dict)
session.last_http_status        # HTTP status code (int)
session.last_response_text      # raw response body (str)

Diagnostic attributes

Attribute	Type	Description
qmgr_name	str	Queue manager name
gateway_qmgr	str \| None	Gateway queue manager (or None)
last_http_status	int	HTTP status code from last command
last_response_text	str	Raw response body from last command
last_response_payload	dict	Parsed response from last command
last_command_payload	dict	Command sent in last request

Transport

See Transport for the transport protocol, response type, and mock transport examples.

API reference

Bases: MQRESTSyncMixin, MQRESTEnsureMixin, MQRESTCommandMixin

Session wrapper for MQ REST admin calls.

Provides MQSC command execution via the IBM MQ runCommandJSON REST endpoint. Attribute mapping between snake_case and native MQSC parameter names is enabled by default.

All MQSC command methods are inherited from :class:~pymqrest.commands.MQRESTCommandMixin — see :doc:/api/commands for the full list.

Attributes:

Name	Type	Description
last_response_payload	dict[str, object] | None	The parsed JSON payload from the most recent command, or None before any command is executed.
last_response_text	str | None	The raw HTTP response body from the most recent command, or None.
last_http_status	int | None	The HTTP status code from the most recent command, or None.
last_command_payload	dict[str, object] | None	The runCommandJSON request payload sent for the most recent command, or None.

Source code in src/pymqrest/session.py

class MQRESTSession(MQRESTSyncMixin, MQRESTEnsureMixin, MQRESTCommandMixin):
    """Session wrapper for MQ REST admin calls.

    Provides MQSC command execution via the IBM MQ ``runCommandJSON``
    REST endpoint. Attribute mapping between ``snake_case`` and native
    MQSC parameter names is enabled by default.

    All MQSC command methods are inherited from
    :class:`~pymqrest.commands.MQRESTCommandMixin` — see
    :doc:`/api/commands` for the full list.

    Attributes:
        last_response_payload: The parsed JSON payload from the most
            recent command, or ``None`` before any command is executed.
        last_response_text: The raw HTTP response body from the most
            recent command, or ``None``.
        last_http_status: The HTTP status code from the most recent
            command, or ``None``.
        last_command_payload: The ``runCommandJSON`` request payload
            sent for the most recent command, or ``None``.

    """

    def __init__(  # noqa: PLR0913
        self,
        rest_base_url: str,
        qmgr_name: str,
        *,
        credentials: Credentials,
        gateway_qmgr: str | None = None,
        verify_tls: bool = True,
        timeout_seconds: float | None = 30.0,
        map_attributes: bool = True,
        mapping_strict: bool = True,
        mapping_overrides: Mapping[str, object] | None = None,
        mapping_overrides_mode: MappingOverrideMode = MappingOverrideMode.MERGE,
        csrf_token: str | None = DEFAULT_CSRF_TOKEN,
        transport: MQRESTTransport | None = None,
    ) -> None:
        """Initialize an MQ REST session.

        Args:
            rest_base_url: Base URL of the MQ REST API
                (e.g. ``"https://localhost:9443/ibmmq/rest/v2"``).
            qmgr_name: Name of the target queue manager.
            credentials: A credential object (:class:`~pymqrest.auth.BasicAuth`,
                :class:`~pymqrest.auth.LTPAAuth`, or
                :class:`~pymqrest.auth.CertificateAuth`).
            gateway_qmgr: Name of the gateway queue manager that routes
                commands to *qmgr_name*. When set, the
                ``ibm-mq-rest-gateway-qmgr`` header is included in
                every request. When ``None`` (default), commands target
                the queue manager directly.
            verify_tls: Whether to verify the server's TLS certificate.
                Set to ``False`` for self-signed certificates.
            timeout_seconds: HTTP request timeout in seconds, or
                ``None`` for no timeout.
            map_attributes: When ``True`` (default), translate attribute
                names between ``snake_case`` and MQSC forms
                automatically.
            mapping_strict: When ``True`` (default), raise
                :class:`~pymqrest.mapping.MappingError` on any
                unrecognised attribute. When ``False``, pass
                unrecognised attributes through unchanged.
            mapping_overrides: Optional overrides for the built-in
                mapping data. Keys must be a subset of
                ``{"commands", "qualifiers"}``. Behaviour depends on
                *mapping_overrides_mode*.
            mapping_overrides_mode: How to apply *mapping_overrides*.
                :attr:`~MappingOverrideMode.MERGE` (default) performs a
                sparse, additive merge.
                :attr:`~MappingOverrideMode.REPLACE` treats the
                overrides as a complete replacement — a ``ValueError``
                is raised if any command or qualifier key from the
                built-in data is missing.
            csrf_token: CSRF token value for the
                ``ibm-mq-rest-csrf-token`` header. Defaults to
                ``"local"``. Set to ``None`` to omit the header.
            transport: Custom :class:`MQRESTTransport` implementation.
                Defaults to :class:`RequestsTransport`.

        Raises:
            MQRESTAuthError: If LTPA login fails at construction time.
            ValueError: If *mapping_overrides* has an invalid structure.

        """
        self._rest_base_url = rest_base_url.rstrip("/")
        self._qmgr_name = qmgr_name
        self._gateway_qmgr = gateway_qmgr
        self._verify_tls = verify_tls
        self._timeout_seconds = timeout_seconds
        self._map_attributes = map_attributes
        self._mapping_strict = mapping_strict
        self._csrf_token = csrf_token
        self._credentials = credentials

        if mapping_overrides is not None:
            validate_mapping_overrides(mapping_overrides)
            if mapping_overrides_mode is MappingOverrideMode.REPLACE:
                validate_mapping_overrides_complete(MAPPING_DATA, mapping_overrides)
                self._mapping_data: dict[str, object] = replace_mapping_data(mapping_overrides)
            else:
                self._mapping_data = merge_mapping_data(MAPPING_DATA, mapping_overrides)
        else:
            self._mapping_data = MAPPING_DATA

        if isinstance(credentials, CertificateAuth) and transport is None:
            cert = (
                (credentials.cert_path, credentials.key_path)
                if credentials.key_path is not None
                else credentials.cert_path
            )
            self._transport: MQRESTTransport = RequestsTransport(client_cert=cert)
        else:
            self._transport = transport or RequestsTransport()

        self._ltpa_cookie_name: str | None = None
        self._ltpa_token: str | None = None
        if isinstance(credentials, LTPAAuth):
            self._ltpa_cookie_name, self._ltpa_token = _perform_ltpa_login(
                self._transport,
                self._rest_base_url,
                credentials,
                csrf_token=self._csrf_token,
                timeout_seconds=self._timeout_seconds,
                verify_tls=self._verify_tls,
            )

        self.last_response_payload: dict[str, object] | None = None
        self.last_response_text: str | None = None
        self.last_http_status: int | None = None
        self.last_command_payload: dict[str, object] | None = None

    @property
    def qmgr_name(self) -> str:
        """The queue manager name this session targets."""
        return self._qmgr_name

    @property
    def gateway_qmgr(self) -> str | None:
        """The gateway queue manager name, or ``None`` for direct access."""
        return self._gateway_qmgr

    def _mqsc_command(
        self,
        *,
        command: str,
        mqsc_qualifier: str,
        name: str | None,
        request_parameters: Mapping[str, object] | None,
        response_parameters: Sequence[str] | None,
        where: str | None = None,
    ) -> list[dict[str, object]]:
        command_upper = command.strip().upper()
        qualifier_upper = mqsc_qualifier.strip().upper()
        normalized_request_parameters = dict(request_parameters or {})
        normalized_response_parameters = _normalize_response_parameters(
            response_parameters,
            is_display=command_upper == "DISPLAY",
        )
        map_attributes = self._map_attributes
        mapping_qualifier = self._resolve_mapping_qualifier(command_upper, qualifier_upper)

        if map_attributes:
            normalized_request_parameters = map_request_attributes(
                mapping_qualifier,
                normalized_request_parameters,
                strict=self._mapping_strict,
                mapping_data=self._mapping_data,
            )
            normalized_response_parameters = self._map_response_parameters(
                command_upper,
                qualifier_upper,
                mapping_qualifier,
                normalized_response_parameters,
            )

        if where is not None and where.strip():
            mapped_where = where
            if map_attributes:
                mapped_where = _map_where_keyword(
                    where,
                    mapping_qualifier,
                    strict=self._mapping_strict,
                    mapping_data=self._mapping_data,
                )
            normalized_request_parameters["WHERE"] = mapped_where

        payload = _build_command_payload(
            command=command_upper,
            qualifier=qualifier_upper,
            name=name,
            request_parameters=normalized_request_parameters,
            response_parameters=normalized_response_parameters,
        )
        self.last_command_payload = dict(payload)
        transport_response = self._transport.post_json(
            self._build_mqsc_url(),
            payload,
            headers=self._build_headers(),
            timeout_seconds=self._timeout_seconds,
            verify_tls=self._verify_tls,
        )
        self.last_http_status = transport_response.status_code
        self.last_response_text = transport_response.text
        response_payload = _parse_response_payload(transport_response.text)
        self.last_response_payload = response_payload
        _raise_for_command_errors(response_payload, transport_response.status_code)

        command_response = _extract_command_response(response_payload)
        parameter_objects: list[dict[str, object]] = []
        for command_response_item in command_response:
            parameters = command_response_item.get("parameters")
            if isinstance(parameters, Mapping):
                parameters_map = cast("Mapping[str, object]", parameters)
                parameter_objects.append(dict(parameters_map))
            else:
                parameter_objects.append({})

        parameter_objects = _flatten_nested_objects(parameter_objects)

        if map_attributes:
            normalized_objects = [_normalize_response_attributes(item) for item in parameter_objects]
            return map_response_list(
                mapping_qualifier,
                normalized_objects,
                strict=self._mapping_strict,
                mapping_data=self._mapping_data,
            )
        return parameter_objects

    def _build_mqsc_url(self) -> str:
        return f"{self._rest_base_url}/admin/action/qmgr/{self._qmgr_name}/mqsc"

    def _build_headers(self) -> dict[str, str]:
        headers: dict[str, str] = {"Accept": "application/json"}
        if isinstance(self._credentials, BasicAuth):
            headers["Authorization"] = _build_basic_auth_header(
                self._credentials.username,
                self._credentials.password,
            )
        elif isinstance(self._credentials, LTPAAuth) and self._ltpa_token is not None:
            headers["Cookie"] = f"{self._ltpa_cookie_name}={self._ltpa_token}"
        if self._csrf_token is not None:
            headers["ibm-mq-rest-csrf-token"] = self._csrf_token
        if self._gateway_qmgr is not None:
            headers[GATEWAY_HEADER] = self._gateway_qmgr
        return headers

    def _map_response_parameters(
        self,
        command: str,
        mqsc_qualifier: str,
        mapping_qualifier: str,
        response_parameters: list[str],
    ) -> list[str]:
        if _is_all_response_parameters(response_parameters):
            return response_parameters
        response_parameter_macros = _get_response_parameter_macros(
            command,
            mqsc_qualifier,
            mapping_data=self._mapping_data,
        )
        macro_lookup = {macro.lower(): macro for macro in response_parameter_macros}
        qualifier_entry = _get_qualifier_entry(mapping_qualifier, mapping_data=self._mapping_data)
        if qualifier_entry is None:
            if self._mapping_strict:
                raise MappingError(_build_unknown_qualifier_issue(mapping_qualifier))
            return response_parameters
        combined_map = _build_snake_to_mqsc_map(qualifier_entry)
        mapped, issues = _map_response_parameter_names(
            response_parameters,
            macro_lookup,
            combined_map,
            mapping_qualifier,
        )
        if self._mapping_strict and issues:
            raise MappingError(issues)
        return mapped

    def _resolve_mapping_qualifier(self, command: str, mqsc_qualifier: str) -> str:
        command_map = _get_command_map(self._mapping_data)
        command_key = f"{command} {mqsc_qualifier}"
        command_definition = command_map.get(command_key)
        if isinstance(command_definition, Mapping):
            command_definition_map = cast("Mapping[str, object]", command_definition)
            qualifier = command_definition_map.get("qualifier")
            if isinstance(qualifier, str):
                return qualifier
        fallback = _DEFAULT_MAPPING_QUALIFIERS.get(mqsc_qualifier)
        if fallback is not None:
            return fallback
        return mqsc_qualifier.lower()

_rest_base_url = rest_base_url.rstrip('/') instance-attribute

_qmgr_name = qmgr_name instance-attribute

_gateway_qmgr = gateway_qmgr instance-attribute

_verify_tls = verify_tls instance-attribute

_timeout_seconds = timeout_seconds instance-attribute

_map_attributes = map_attributes instance-attribute

_mapping_strict = mapping_strict instance-attribute

_csrf_token = csrf_token instance-attribute

_credentials = credentials instance-attribute

_mapping_data = replace_mapping_data(mapping_overrides) instance-attribute

_transport = RequestsTransport(client_cert=cert) instance-attribute

_ltpa_cookie_name = None instance-attribute

_ltpa_token = None instance-attribute

last_response_payload = None instance-attribute

last_response_text = None instance-attribute

last_http_status = None instance-attribute

last_command_payload = None instance-attribute

qmgr_name property

The queue manager name this session targets.

gateway_qmgr property

The gateway queue manager name, or None for direct access.

__init__(rest_base_url, qmgr_name, *, credentials, gateway_qmgr=None, verify_tls=True, timeout_seconds=30.0, map_attributes=True, mapping_strict=True, mapping_overrides=None, mapping_overrides_mode=MappingOverrideMode.MERGE, csrf_token=DEFAULT_CSRF_TOKEN, transport=None)

Initialize an MQ REST session.

Parameters:

Name	Type	Description	Default
rest_base_url	str	Base URL of the MQ REST API (e.g. "https://localhost:9443/ibmmq/rest/v2").	required
qmgr_name	str	Name of the target queue manager.	required
credentials	Credentials	A credential object (:class:~pymqrest.auth.BasicAuth, :class:~pymqrest.auth.LTPAAuth, or :class:~pymqrest.auth.CertificateAuth).	required
gateway_qmgr	str | None	Name of the gateway queue manager that routes commands to qmgr_name. When set, the ibm-mq-rest-gateway-qmgr header is included in every request. When None (default), commands target the queue manager directly.	None
verify_tls	bool	Whether to verify the server's TLS certificate. Set to False for self-signed certificates.	True
timeout_seconds	float | None	HTTP request timeout in seconds, or None for no timeout.	30.0
map_attributes	bool	When True (default), translate attribute names between snake_case and MQSC forms automatically.	True
mapping_strict	bool	When True (default), raise :class:~pymqrest.mapping.MappingError on any unrecognised attribute. When False, pass unrecognised attributes through unchanged.	True
mapping_overrides	Mapping[str, object] | None	Optional overrides for the built-in mapping data. Keys must be a subset of {"commands", "qualifiers"}. Behaviour depends on mapping_overrides_mode.	None
mapping_overrides_mode	MappingOverrideMode	How to apply mapping_overrides. :attr:~MappingOverrideMode.MERGE (default) performs a sparse, additive merge. :attr:~MappingOverrideMode.REPLACE treats the overrides as a complete replacement — a ValueError is raised if any command or qualifier key from the built-in data is missing.	MERGE
csrf_token	str | None	CSRF token value for the ibm-mq-rest-csrf-token header. Defaults to "local". Set to None to omit the header.	DEFAULT_CSRF_TOKEN
transport	MQRESTTransport | None	Custom :class:MQRESTTransport implementation. Defaults to :class:RequestsTransport.	None

Raises:

Type	Description
MQRESTAuthError	If LTPA login fails at construction time.
ValueError	If mapping_overrides has an invalid structure.

Source code in src/pymqrest/session.py

def __init__(  # noqa: PLR0913
    self,
    rest_base_url: str,
    qmgr_name: str,
    *,
    credentials: Credentials,
    gateway_qmgr: str | None = None,
    verify_tls: bool = True,
    timeout_seconds: float | None = 30.0,
    map_attributes: bool = True,
    mapping_strict: bool = True,
    mapping_overrides: Mapping[str, object] | None = None,
    mapping_overrides_mode: MappingOverrideMode = MappingOverrideMode.MERGE,
    csrf_token: str | None = DEFAULT_CSRF_TOKEN,
    transport: MQRESTTransport | None = None,
) -> None:
    """Initialize an MQ REST session.

    Args:
        rest_base_url: Base URL of the MQ REST API
            (e.g. ``"https://localhost:9443/ibmmq/rest/v2"``).
        qmgr_name: Name of the target queue manager.
        credentials: A credential object (:class:`~pymqrest.auth.BasicAuth`,
            :class:`~pymqrest.auth.LTPAAuth`, or
            :class:`~pymqrest.auth.CertificateAuth`).
        gateway_qmgr: Name of the gateway queue manager that routes
            commands to *qmgr_name*. When set, the
            ``ibm-mq-rest-gateway-qmgr`` header is included in
            every request. When ``None`` (default), commands target
            the queue manager directly.
        verify_tls: Whether to verify the server's TLS certificate.
            Set to ``False`` for self-signed certificates.
        timeout_seconds: HTTP request timeout in seconds, or
            ``None`` for no timeout.
        map_attributes: When ``True`` (default), translate attribute
            names between ``snake_case`` and MQSC forms
            automatically.
        mapping_strict: When ``True`` (default), raise
            :class:`~pymqrest.mapping.MappingError` on any
            unrecognised attribute. When ``False``, pass
            unrecognised attributes through unchanged.
        mapping_overrides: Optional overrides for the built-in
            mapping data. Keys must be a subset of
            ``{"commands", "qualifiers"}``. Behaviour depends on
            *mapping_overrides_mode*.
        mapping_overrides_mode: How to apply *mapping_overrides*.
            :attr:`~MappingOverrideMode.MERGE` (default) performs a
            sparse, additive merge.
            :attr:`~MappingOverrideMode.REPLACE` treats the
            overrides as a complete replacement — a ``ValueError``
            is raised if any command or qualifier key from the
            built-in data is missing.
        csrf_token: CSRF token value for the
            ``ibm-mq-rest-csrf-token`` header. Defaults to
            ``"local"``. Set to ``None`` to omit the header.
        transport: Custom :class:`MQRESTTransport` implementation.
            Defaults to :class:`RequestsTransport`.

    Raises:
        MQRESTAuthError: If LTPA login fails at construction time.
        ValueError: If *mapping_overrides* has an invalid structure.

    """
    self._rest_base_url = rest_base_url.rstrip("/")
    self._qmgr_name = qmgr_name
    self._gateway_qmgr = gateway_qmgr
    self._verify_tls = verify_tls
    self._timeout_seconds = timeout_seconds
    self._map_attributes = map_attributes
    self._mapping_strict = mapping_strict
    self._csrf_token = csrf_token
    self._credentials = credentials

    if mapping_overrides is not None:
        validate_mapping_overrides(mapping_overrides)
        if mapping_overrides_mode is MappingOverrideMode.REPLACE:
            validate_mapping_overrides_complete(MAPPING_DATA, mapping_overrides)
            self._mapping_data: dict[str, object] = replace_mapping_data(mapping_overrides)
        else:
            self._mapping_data = merge_mapping_data(MAPPING_DATA, mapping_overrides)
    else:
        self._mapping_data = MAPPING_DATA

    if isinstance(credentials, CertificateAuth) and transport is None:
        cert = (
            (credentials.cert_path, credentials.key_path)
            if credentials.key_path is not None
            else credentials.cert_path
        )
        self._transport: MQRESTTransport = RequestsTransport(client_cert=cert)
    else:
        self._transport = transport or RequestsTransport()

    self._ltpa_cookie_name: str | None = None
    self._ltpa_token: str | None = None
    if isinstance(credentials, LTPAAuth):
        self._ltpa_cookie_name, self._ltpa_token = _perform_ltpa_login(
            self._transport,
            self._rest_base_url,
            credentials,
            csrf_token=self._csrf_token,
            timeout_seconds=self._timeout_seconds,
            verify_tls=self._verify_tls,
        )

    self.last_response_payload: dict[str, object] | None = None
    self.last_response_text: str | None = None
    self.last_http_status: int | None = None
    self.last_command_payload: dict[str, object] | None = None

_mqsc_command(*, command, mqsc_qualifier, name, request_parameters, response_parameters, where=None)

Source code in src/pymqrest/session.py

def _mqsc_command(
    self,
    *,
    command: str,
    mqsc_qualifier: str,
    name: str | None,
    request_parameters: Mapping[str, object] | None,
    response_parameters: Sequence[str] | None,
    where: str | None = None,
) -> list[dict[str, object]]:
    command_upper = command.strip().upper()
    qualifier_upper = mqsc_qualifier.strip().upper()
    normalized_request_parameters = dict(request_parameters or {})
    normalized_response_parameters = _normalize_response_parameters(
        response_parameters,
        is_display=command_upper == "DISPLAY",
    )
    map_attributes = self._map_attributes
    mapping_qualifier = self._resolve_mapping_qualifier(command_upper, qualifier_upper)

    if map_attributes:
        normalized_request_parameters = map_request_attributes(
            mapping_qualifier,
            normalized_request_parameters,
            strict=self._mapping_strict,
            mapping_data=self._mapping_data,
        )
        normalized_response_parameters = self._map_response_parameters(
            command_upper,
            qualifier_upper,
            mapping_qualifier,
            normalized_response_parameters,
        )

    if where is not None and where.strip():
        mapped_where = where
        if map_attributes:
            mapped_where = _map_where_keyword(
                where,
                mapping_qualifier,
                strict=self._mapping_strict,
                mapping_data=self._mapping_data,
            )
        normalized_request_parameters["WHERE"] = mapped_where

    payload = _build_command_payload(
        command=command_upper,
        qualifier=qualifier_upper,
        name=name,
        request_parameters=normalized_request_parameters,
        response_parameters=normalized_response_parameters,
    )
    self.last_command_payload = dict(payload)
    transport_response = self._transport.post_json(
        self._build_mqsc_url(),
        payload,
        headers=self._build_headers(),
        timeout_seconds=self._timeout_seconds,
        verify_tls=self._verify_tls,
    )
    self.last_http_status = transport_response.status_code
    self.last_response_text = transport_response.text
    response_payload = _parse_response_payload(transport_response.text)
    self.last_response_payload = response_payload
    _raise_for_command_errors(response_payload, transport_response.status_code)

    command_response = _extract_command_response(response_payload)
    parameter_objects: list[dict[str, object]] = []
    for command_response_item in command_response:
        parameters = command_response_item.get("parameters")
        if isinstance(parameters, Mapping):
            parameters_map = cast("Mapping[str, object]", parameters)
            parameter_objects.append(dict(parameters_map))
        else:
            parameter_objects.append({})

    parameter_objects = _flatten_nested_objects(parameter_objects)

    if map_attributes:
        normalized_objects = [_normalize_response_attributes(item) for item in parameter_objects]
        return map_response_list(
            mapping_qualifier,
            normalized_objects,
            strict=self._mapping_strict,
            mapping_data=self._mapping_data,
        )
    return parameter_objects

_build_mqsc_url()

Source code in src/pymqrest/session.py

def _build_mqsc_url(self) -> str:
    return f"{self._rest_base_url}/admin/action/qmgr/{self._qmgr_name}/mqsc"

_build_headers()

Source code in src/pymqrest/session.py

def _build_headers(self) -> dict[str, str]:
    headers: dict[str, str] = {"Accept": "application/json"}
    if isinstance(self._credentials, BasicAuth):
        headers["Authorization"] = _build_basic_auth_header(
            self._credentials.username,
            self._credentials.password,
        )
    elif isinstance(self._credentials, LTPAAuth) and self._ltpa_token is not None:
        headers["Cookie"] = f"{self._ltpa_cookie_name}={self._ltpa_token}"
    if self._csrf_token is not None:
        headers["ibm-mq-rest-csrf-token"] = self._csrf_token
    if self._gateway_qmgr is not None:
        headers[GATEWAY_HEADER] = self._gateway_qmgr
    return headers

_map_response_parameters(command, mqsc_qualifier, mapping_qualifier, response_parameters)

Source code in src/pymqrest/session.py

def _map_response_parameters(
    self,
    command: str,
    mqsc_qualifier: str,
    mapping_qualifier: str,
    response_parameters: list[str],
) -> list[str]:
    if _is_all_response_parameters(response_parameters):
        return response_parameters
    response_parameter_macros = _get_response_parameter_macros(
        command,
        mqsc_qualifier,
        mapping_data=self._mapping_data,
    )
    macro_lookup = {macro.lower(): macro for macro in response_parameter_macros}
    qualifier_entry = _get_qualifier_entry(mapping_qualifier, mapping_data=self._mapping_data)
    if qualifier_entry is None:
        if self._mapping_strict:
            raise MappingError(_build_unknown_qualifier_issue(mapping_qualifier))
        return response_parameters
    combined_map = _build_snake_to_mqsc_map(qualifier_entry)
    mapped, issues = _map_response_parameter_names(
        response_parameters,
        macro_lookup,
        combined_map,
        mapping_qualifier,
    )
    if self._mapping_strict and issues:
        raise MappingError(issues)
    return mapped

_resolve_mapping_qualifier(command, mqsc_qualifier)

Source code in src/pymqrest/session.py

def _resolve_mapping_qualifier(self, command: str, mqsc_qualifier: str) -> str:
    command_map = _get_command_map(self._mapping_data)
    command_key = f"{command} {mqsc_qualifier}"
    command_definition = command_map.get(command_key)
    if isinstance(command_definition, Mapping):
        command_definition_map = cast("Mapping[str, object]", command_definition)
        qualifier = command_definition_map.get("qualifier")
        if isinstance(qualifier, str):
            return qualifier
    fallback = _DEFAULT_MAPPING_QUALIFIERS.get(mqsc_qualifier)
    if fallback is not None:
        return fallback
    return mqsc_qualifier.lower()