Ensure

Overview

The ensure module provides the return types for the 16 idempotent ensure methods on MQRESTSession. These methods implement a declarative upsert pattern: DEFINE if the object does not exist, ALTER only attributes that differ, or no-op if the object already matches the desired state.

EnsureAction

An enum indicating the action taken by an ensure method:

class EnsureAction(Enum):
    CREATED    = "created"     # Object did not exist; DEFINE was issued
    UPDATED    = "updated"     # Object existed but attributes differed; ALTER was issued
    UNCHANGED  = "unchanged"   # Object already matched the desired state

EnsureResult

A named tuple containing the action taken and the list of attribute names that triggered the change (if any):

class EnsureResult(NamedTuple):
    action: EnsureAction       # What happened
    changed: list[str]         # Attribute names that differed (empty for CREATED/UNCHANGED)

Attribute	Type	Description
action	EnsureAction	What happened: CREATED, UPDATED, or UNCHANGED
changed	list[str]	Attribute names that triggered an ALTER (in the caller's namespace)

Usage

result = session.ensure_qlocal("MY.QUEUE",
    request_parameters={"max_queue_depth": 50000, "description": "App queue"})

match result.action:
    case EnsureAction.CREATED:
        print("Queue created")
    case EnsureAction.UPDATED:
        print(f"Changed: {result.changed}")
    case EnsureAction.UNCHANGED:
        print("Already correct")

See Ensure Methods for the full conceptual overview, comparison logic, and the complete list of available ensure methods.

API reference

Bases: Enum

Action taken by an ensure operation.

Attributes:

Name	Type	Description
CREATED		The object did not exist and was defined.
UPDATED		The object existed but attributes differed and were altered.
UNCHANGED		The object existed and all specified attributes already matched.

Source code in src/pymqrest/ensure.py

class EnsureAction(enum.Enum):
    """Action taken by an ensure operation.

    Attributes:
        CREATED: The object did not exist and was defined.
        UPDATED: The object existed but attributes differed and were altered.
        UNCHANGED: The object existed and all specified attributes already matched.

    """

    CREATED = "created"
    UPDATED = "updated"
    UNCHANGED = "unchanged"

CREATED = 'created' class-attribute instance-attribute

UPDATED = 'updated' class-attribute instance-attribute

UNCHANGED = 'unchanged' class-attribute instance-attribute

Result of an ensure operation.

Attributes:

Name	Type	Description
action	EnsureAction	The :class:EnsureAction indicating what happened.
changed	tuple[str, ...]	Attribute names that triggered the ALTER, in the caller's namespace (snake_case when mapping is enabled, MQSC when disabled). Empty for CREATED and UNCHANGED actions.

Source code in src/pymqrest/ensure.py

@dataclass(frozen=True)
class EnsureResult:
    """Result of an ensure operation.

    Attributes:
        action: The :class:`EnsureAction` indicating what happened.
        changed: Attribute names that triggered the ALTER, in the caller's
            namespace (snake_case when mapping is enabled, MQSC when
            disabled).  Empty for CREATED and UNCHANGED actions.

    """

    action: EnsureAction
    changed: tuple[str, ...] = ()

action instance-attribute

changed = () class-attribute instance-attribute

__init__(action, changed=())

Mixin providing idempotent ensure methods for MQ objects.

Each ensure_* method implements a declarative upsert pattern: DEFINE when the object does not exist, ALTER only when specified attributes differ, and no-op when they already match — preserving ALTDATE/ALTTIME for unchanged objects.

Source code in src/pymqrest/ensure.py

class MQRESTEnsureMixin:
    """Mixin providing idempotent ensure methods for MQ objects.

    Each ``ensure_*`` method implements a declarative upsert pattern:
    DEFINE when the object does not exist, ALTER only when specified
    attributes differ, and no-op when they already match — preserving
    ``ALTDATE``/``ALTTIME`` for unchanged objects.
    """

    def _mqsc_command(  # noqa: PLR0913
        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]]:
        raise NotImplementedError  # pragma: no cover

    def _ensure_object(
        self,
        *,
        name: str,
        request_parameters: Mapping[str, object] | None,
        display_qualifier: str,
        define_qualifier: str,
        alter_qualifier: str,
    ) -> EnsureResult:
        """Core ensure logic shared by all ``ensure_*`` methods.

        Args:
            name: MQ object name.
            request_parameters: Desired attributes to assert/set.
            display_qualifier: MQSC qualifier for the DISPLAY command.
            define_qualifier: MQSC qualifier for the DEFINE command.
            alter_qualifier: MQSC qualifier for the ALTER command.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        try:
            current_objects = self._mqsc_command(
                command="DISPLAY",
                mqsc_qualifier=display_qualifier,
                name=name,
                request_parameters=None,
                response_parameters=["all"],
            )
        except MQRESTCommandError:
            current_objects = []

        params = dict(request_parameters) if request_parameters else {}

        if not current_objects:
            self._mqsc_command(
                command="DEFINE",
                mqsc_qualifier=define_qualifier,
                name=name,
                request_parameters=params or None,
                response_parameters=None,
            )
            return EnsureResult(EnsureAction.CREATED)

        if not params:
            return EnsureResult(EnsureAction.UNCHANGED)

        current = current_objects[0]
        changed: dict[str, object] = {}
        for key, desired_value in params.items():
            current_value = current.get(key)
            if not _values_match(desired_value, current_value):
                changed[key] = desired_value

        if not changed:
            return EnsureResult(EnsureAction.UNCHANGED)

        self._mqsc_command(
            command="ALTER",
            mqsc_qualifier=alter_qualifier,
            name=name,
            request_parameters=changed,
            response_parameters=None,
        )
        return EnsureResult(EnsureAction.UPDATED, changed=tuple(changed.keys()))

    def ensure_qmgr(
        self,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure the queue manager has the specified attributes.

        Unlike other ensure methods, the queue manager always exists and
        cannot be defined or deleted.  This method compares the requested
        attributes against the current state and issues ``ALTER QMGR``
        only when values differ.  Returns ``UPDATED`` or ``UNCHANGED``
        (never ``CREATED``).

        Args:
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        params = dict(request_parameters) if request_parameters else {}
        if not params:
            return EnsureResult(EnsureAction.UNCHANGED)

        current_objects = self._mqsc_command(
            command="DISPLAY",
            mqsc_qualifier="QMGR",
            name=None,
            request_parameters=None,
            response_parameters=["all"],
        )

        current = current_objects[0] if current_objects else {}
        changed: dict[str, object] = {}
        for key, desired_value in params.items():
            current_value = current.get(key)
            if not _values_match(desired_value, current_value):
                changed[key] = desired_value

        if not changed:
            return EnsureResult(EnsureAction.UNCHANGED)

        self._mqsc_command(
            command="ALTER",
            mqsc_qualifier="QMGR",
            name=None,
            request_parameters=changed,
            response_parameters=None,
        )
        return EnsureResult(EnsureAction.UPDATED, changed=tuple(changed.keys()))

    def ensure_qlocal(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a local queue exists with the specified attributes.

        Args:
            name: Queue name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="QUEUE",
            define_qualifier="QLOCAL",
            alter_qualifier="QLOCAL",
        )

    def ensure_qremote(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a remote queue exists with the specified attributes.

        Args:
            name: Queue name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="QUEUE",
            define_qualifier="QREMOTE",
            alter_qualifier="QREMOTE",
        )

    def ensure_qalias(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure an alias queue exists with the specified attributes.

        Args:
            name: Queue name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="QUEUE",
            define_qualifier="QALIAS",
            alter_qualifier="QALIAS",
        )

    def ensure_qmodel(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a model queue exists with the specified attributes.

        Args:
            name: Queue name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="QUEUE",
            define_qualifier="QMODEL",
            alter_qualifier="QMODEL",
        )

    def ensure_channel(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a channel exists with the specified attributes.

        Args:
            name: Channel name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="CHANNEL",
            define_qualifier="CHANNEL",
            alter_qualifier="CHANNEL",
        )

    def ensure_authinfo(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure an authentication information object exists with the specified attributes.

        Args:
            name: Authentication information object name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="AUTHINFO",
            define_qualifier="AUTHINFO",
            alter_qualifier="AUTHINFO",
        )

    def ensure_listener(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a listener exists with the specified attributes.

        Args:
            name: Listener name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="LISTENER",
            define_qualifier="LISTENER",
            alter_qualifier="LISTENER",
        )

    def ensure_namelist(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a namelist exists with the specified attributes.

        Args:
            name: Namelist name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="NAMELIST",
            define_qualifier="NAMELIST",
            alter_qualifier="NAMELIST",
        )

    def ensure_process(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a process exists with the specified attributes.

        Args:
            name: Process name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="PROCESS",
            define_qualifier="PROCESS",
            alter_qualifier="PROCESS",
        )

    def ensure_service(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a service exists with the specified attributes.

        Args:
            name: Service name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="SERVICE",
            define_qualifier="SERVICE",
            alter_qualifier="SERVICE",
        )

    def ensure_topic(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a topic exists with the specified attributes.

        Args:
            name: Topic name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="TOPIC",
            define_qualifier="TOPIC",
            alter_qualifier="TOPIC",
        )

    def ensure_sub(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a subscription exists with the specified attributes.

        Args:
            name: Subscription name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="SUB",
            define_qualifier="SUB",
            alter_qualifier="SUB",
        )

    def ensure_stgclass(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a storage class exists with the specified attributes.

        Args:
            name: Storage class name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="STGCLASS",
            define_qualifier="STGCLASS",
            alter_qualifier="STGCLASS",
        )

    def ensure_comminfo(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a communication information object exists with the specified attributes.

        Args:
            name: Communication information object name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="COMMINFO",
            define_qualifier="COMMINFO",
            alter_qualifier="COMMINFO",
        )

    def ensure_cfstruct(
        self,
        name: str,
        request_parameters: Mapping[str, object] | None = None,
    ) -> EnsureResult:
        """Ensure a CF structure exists with the specified attributes.

        Args:
            name: CF structure name.
            request_parameters: Desired attributes to assert/set.

        Returns:
            The :class:`EnsureResult` indicating what action was taken.

        """
        return self._ensure_object(
            name=name,
            request_parameters=request_parameters,
            display_qualifier="CFSTRUCT",
            define_qualifier="CFSTRUCT",
            alter_qualifier="CFSTRUCT",
        )

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

Source code in src/pymqrest/ensure.py

def _mqsc_command(  # noqa: PLR0913
    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]]:
    raise NotImplementedError  # pragma: no cover

_ensure_object(*, name, request_parameters, display_qualifier, define_qualifier, alter_qualifier)

Core ensure logic shared by all ensure_* methods.

Parameters:

Name	Type	Description	Default
name	str	MQ object name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	required
display_qualifier	str	MQSC qualifier for the DISPLAY command.	required
define_qualifier	str	MQSC qualifier for the DEFINE command.	required
alter_qualifier	str	MQSC qualifier for the ALTER command.	required

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def _ensure_object(
    self,
    *,
    name: str,
    request_parameters: Mapping[str, object] | None,
    display_qualifier: str,
    define_qualifier: str,
    alter_qualifier: str,
) -> EnsureResult:
    """Core ensure logic shared by all ``ensure_*`` methods.

    Args:
        name: MQ object name.
        request_parameters: Desired attributes to assert/set.
        display_qualifier: MQSC qualifier for the DISPLAY command.
        define_qualifier: MQSC qualifier for the DEFINE command.
        alter_qualifier: MQSC qualifier for the ALTER command.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    try:
        current_objects = self._mqsc_command(
            command="DISPLAY",
            mqsc_qualifier=display_qualifier,
            name=name,
            request_parameters=None,
            response_parameters=["all"],
        )
    except MQRESTCommandError:
        current_objects = []

    params = dict(request_parameters) if request_parameters else {}

    if not current_objects:
        self._mqsc_command(
            command="DEFINE",
            mqsc_qualifier=define_qualifier,
            name=name,
            request_parameters=params or None,
            response_parameters=None,
        )
        return EnsureResult(EnsureAction.CREATED)

    if not params:
        return EnsureResult(EnsureAction.UNCHANGED)

    current = current_objects[0]
    changed: dict[str, object] = {}
    for key, desired_value in params.items():
        current_value = current.get(key)
        if not _values_match(desired_value, current_value):
            changed[key] = desired_value

    if not changed:
        return EnsureResult(EnsureAction.UNCHANGED)

    self._mqsc_command(
        command="ALTER",
        mqsc_qualifier=alter_qualifier,
        name=name,
        request_parameters=changed,
        response_parameters=None,
    )
    return EnsureResult(EnsureAction.UPDATED, changed=tuple(changed.keys()))

ensure_qmgr(request_parameters=None)

Ensure the queue manager has the specified attributes.

Unlike other ensure methods, the queue manager always exists and cannot be defined or deleted. This method compares the requested attributes against the current state and issues ALTER QMGR only when values differ. Returns UPDATED or UNCHANGED (never CREATED).

Parameters:

Name	Type	Description	Default
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_qmgr(
    self,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure the queue manager has the specified attributes.

    Unlike other ensure methods, the queue manager always exists and
    cannot be defined or deleted.  This method compares the requested
    attributes against the current state and issues ``ALTER QMGR``
    only when values differ.  Returns ``UPDATED`` or ``UNCHANGED``
    (never ``CREATED``).

    Args:
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    params = dict(request_parameters) if request_parameters else {}
    if not params:
        return EnsureResult(EnsureAction.UNCHANGED)

    current_objects = self._mqsc_command(
        command="DISPLAY",
        mqsc_qualifier="QMGR",
        name=None,
        request_parameters=None,
        response_parameters=["all"],
    )

    current = current_objects[0] if current_objects else {}
    changed: dict[str, object] = {}
    for key, desired_value in params.items():
        current_value = current.get(key)
        if not _values_match(desired_value, current_value):
            changed[key] = desired_value

    if not changed:
        return EnsureResult(EnsureAction.UNCHANGED)

    self._mqsc_command(
        command="ALTER",
        mqsc_qualifier="QMGR",
        name=None,
        request_parameters=changed,
        response_parameters=None,
    )
    return EnsureResult(EnsureAction.UPDATED, changed=tuple(changed.keys()))

ensure_qlocal(name, request_parameters=None)

Ensure a local queue exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Queue name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_qlocal(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a local queue exists with the specified attributes.

    Args:
        name: Queue name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="QUEUE",
        define_qualifier="QLOCAL",
        alter_qualifier="QLOCAL",
    )

ensure_qremote(name, request_parameters=None)

Ensure a remote queue exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Queue name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_qremote(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a remote queue exists with the specified attributes.

    Args:
        name: Queue name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="QUEUE",
        define_qualifier="QREMOTE",
        alter_qualifier="QREMOTE",
    )

ensure_qalias(name, request_parameters=None)

Ensure an alias queue exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Queue name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_qalias(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure an alias queue exists with the specified attributes.

    Args:
        name: Queue name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="QUEUE",
        define_qualifier="QALIAS",
        alter_qualifier="QALIAS",
    )

ensure_qmodel(name, request_parameters=None)

Ensure a model queue exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Queue name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_qmodel(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a model queue exists with the specified attributes.

    Args:
        name: Queue name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="QUEUE",
        define_qualifier="QMODEL",
        alter_qualifier="QMODEL",
    )

ensure_channel(name, request_parameters=None)

Ensure a channel exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Channel name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_channel(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a channel exists with the specified attributes.

    Args:
        name: Channel name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="CHANNEL",
        define_qualifier="CHANNEL",
        alter_qualifier="CHANNEL",
    )

ensure_authinfo(name, request_parameters=None)

Ensure an authentication information object exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Authentication information object name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_authinfo(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure an authentication information object exists with the specified attributes.

    Args:
        name: Authentication information object name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="AUTHINFO",
        define_qualifier="AUTHINFO",
        alter_qualifier="AUTHINFO",
    )

ensure_listener(name, request_parameters=None)

Ensure a listener exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Listener name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_listener(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a listener exists with the specified attributes.

    Args:
        name: Listener name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="LISTENER",
        define_qualifier="LISTENER",
        alter_qualifier="LISTENER",
    )

ensure_namelist(name, request_parameters=None)

Ensure a namelist exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Namelist name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_namelist(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a namelist exists with the specified attributes.

    Args:
        name: Namelist name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="NAMELIST",
        define_qualifier="NAMELIST",
        alter_qualifier="NAMELIST",
    )

ensure_process(name, request_parameters=None)

Ensure a process exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Process name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_process(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a process exists with the specified attributes.

    Args:
        name: Process name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="PROCESS",
        define_qualifier="PROCESS",
        alter_qualifier="PROCESS",
    )

ensure_service(name, request_parameters=None)

Ensure a service exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Service name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_service(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a service exists with the specified attributes.

    Args:
        name: Service name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="SERVICE",
        define_qualifier="SERVICE",
        alter_qualifier="SERVICE",
    )

ensure_topic(name, request_parameters=None)

Ensure a topic exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Topic name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_topic(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a topic exists with the specified attributes.

    Args:
        name: Topic name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="TOPIC",
        define_qualifier="TOPIC",
        alter_qualifier="TOPIC",
    )

ensure_sub(name, request_parameters=None)

Ensure a subscription exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Subscription name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_sub(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a subscription exists with the specified attributes.

    Args:
        name: Subscription name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="SUB",
        define_qualifier="SUB",
        alter_qualifier="SUB",
    )

ensure_stgclass(name, request_parameters=None)

Ensure a storage class exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Storage class name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_stgclass(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a storage class exists with the specified attributes.

    Args:
        name: Storage class name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="STGCLASS",
        define_qualifier="STGCLASS",
        alter_qualifier="STGCLASS",
    )

ensure_comminfo(name, request_parameters=None)

Ensure a communication information object exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	Communication information object name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_comminfo(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a communication information object exists with the specified attributes.

    Args:
        name: Communication information object name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="COMMINFO",
        define_qualifier="COMMINFO",
        alter_qualifier="COMMINFO",
    )

ensure_cfstruct(name, request_parameters=None)

Ensure a CF structure exists with the specified attributes.

Parameters:

Name	Type	Description	Default
name	str	CF structure name.	required
request_parameters	Mapping[str, object] | None	Desired attributes to assert/set.	None

Returns:

Name	Type	Description
The	EnsureResult	class:EnsureResult indicating what action was taken.

Source code in src/pymqrest/ensure.py

def ensure_cfstruct(
    self,
    name: str,
    request_parameters: Mapping[str, object] | None = None,
) -> EnsureResult:
    """Ensure a CF structure exists with the specified attributes.

    Args:
        name: CF structure name.
        request_parameters: Desired attributes to assert/set.

    Returns:
        The :class:`EnsureResult` indicating what action was taken.

    """
    return self._ensure_object(
        name=name,
        request_parameters=request_parameters,
        display_qualifier="CFSTRUCT",
        define_qualifier="CFSTRUCT",
        alter_qualifier="CFSTRUCT",
    )