API reference

rmqaio

Number module-attribute

Number = int | float

Numeric type alias for integers or floats.

Config dataclass

Configuration for rmqaio library.

Attributes:

• log_sanitize (bool) –

  Flag indicating whether to sanitize logs by replacing user data with <hidden>.

• log_data_truncate_size (int) –

  Maximum size of data to log before truncation.

Source code in rmqaio/rmqaio.py

@dataclass
class Config:
    """
    Configuration for rmqaio library.

    Attributes:
        log_sanitize: Flag indicating whether to sanitize logs by replacing
            user data with `<hidden>`.
        log_data_truncate_size: Maximum size of data to log before truncation.
    """

    log_sanitize: bool = field(default_factory=lambda: _env_var_as_bool("RMQAIO_LOG_SANITIZE", True))
    """Flag indicating whether to sanitize logs by replacing user data with `<hidden>`."""

    log_data_truncate_size: int = field(default_factory=lambda: _env_var_as_int("RMQAIO_LOG_DATA_TRUNCATE_SIZE", 10000))
    """Maximum size of data to log before truncation."""

log_sanitize class-attribute instance-attribute

log_sanitize: bool = field(
    default_factory=lambda: _env_var_as_bool(
        "RMQAIO_LOG_SANITIZE", True
    )
)

Flag indicating whether to sanitize logs by replacing user data with <hidden>.

log_data_truncate_size class-attribute instance-attribute

log_data_truncate_size: int = field(
    default_factory=lambda: _env_var_as_int(
        "RMQAIO_LOG_DATA_TRUNCATE_SIZE", 10000
    )
)

Maximum size of data to log before truncation.

Repeat

Represents a fixed delay value for retry operations.

Supports hashing and equality comparison, making it suitable for use as a dictionary key or in sets.

Attributes:

• value –

  The value to repeat.

Examples:

>>> repeat = Repeat(5)
>>> for delay in repeat:
...     print(delay)

Source code in rmqaio/rmqaio.py

class Repeat:
    """
    Represents a fixed delay value for retry operations.

    Supports hashing and equality comparison, making it suitable for use as
    a dictionary key or in sets.

    Attributes:
        value: The value to repeat.

    Examples:
        >>> repeat = Repeat(5)
        >>> for delay in repeat:
        ...     print(delay)
    """

    def __init__(self, value: Number):
        """
        Initialize repeat with a value.

        Args:
            value: The value to repeat.
        """
        self.value = value

    def __str__(self):
        return f"{self.__class__.__name__}({self.value})"

    def __iter__(self):
        return iter(itertools.repeat(self.value))

    def __hash__(self):
        return hash((self.__class__, float(self.value)))

    def __eq__(self, other):
        return other.__class__ == Repeat and self.value == other.value

    def __ne__(self, other):
        return other.__class__ != Repeat or self.value != other.value

__init__

__init__(value: Number)

Initialize repeat with a value.

Parameters:

• value (Number) –

  The value to repeat.

Source code in rmqaio/rmqaio.py

def __init__(self, value: Number):
    """
    Initialize repeat with a value.

    Args:
        value: The value to repeat.
    """
    self.value = value

RetryPolicy dataclass

Defines retry policy for handling exceptions and retries.

Attributes:

• delays (Delay) –

  A sequence of delays (in seconds) for attempts.

• exc_filter (tuple[type[Exception], ...] | Callable[[Exception], bool]) –

  A tuple of exception types or a callable function to filter which exceptions should be retried.

Source code in rmqaio/rmqaio.py

@dataclass(slots=True, frozen=True)
class RetryPolicy:
    """
    Defines retry policy for handling exceptions and retries.

    Attributes:
        delays: A sequence of delays (in seconds) for attempts.
        exc_filter: A tuple of exception types or a callable function to
            filter which exceptions should be retried.
    """

    delays: Delay = field(default_factory=lambda: Repeat(5))
    exc_filter: tuple[type[Exception], ...] | Callable[[Exception], bool] = (
        asyncio.TimeoutError,
        ConnectionError,
        aiormq.exceptions.AMQPConnectionError,
    )

    def __hash__(self):
        return hash(
            (
                self.__class__,
                self.delays if isinstance(self.delays, Repeat) else tuple(self.delays or ()),
                self.exc_filter,
            )
        )

    def is_retryable(self, e: Exception) -> bool:
        """
        Determine if an exception is retryable.

        Args:
            e: The exception to check.

        Returns:
            `True` if the exception is retryable, `False` otherwise.
            Delegates to `exc_filter` callable if provided.
        """
        if callable(self.exc_filter):
            return self.exc_filter(e)

        return isinstance(e, self.exc_filter)

is_retryable

is_retryable(e: Exception) -> bool

Determine if an exception is retryable.

Parameters:

• e (Exception) –

  The exception to check.

Returns:

• bool –

  True if the exception is retryable, False otherwise.

• bool –

  Delegates to exc_filter callable if provided.

Source code in rmqaio/rmqaio.py

def is_retryable(self, e: Exception) -> bool:
    """
    Determine if an exception is retryable.

    Args:
        e: The exception to check.

    Returns:
        `True` if the exception is retryable, `False` otherwise.
        Delegates to `exc_filter` callable if provided.
    """
    if callable(self.exc_filter):
        return self.exc_filter(e)

    return isinstance(e, self.exc_filter)

ConnectionState

Bases: StrEnum

Enum for representing the state of a connection.

Source code in rmqaio/rmqaio.py

class ConnectionState(StrEnum):
    """Enum for representing the state of a connection."""

    INITIAL = "initial"
    """Connection is created and ready to open."""

    CONNECTING = "connecting"
    """Connection is in the process of being established."""

    CONNECTED = "connected"
    """Connection is established and operational."""

    REFRESHING = "refreshing"
    """Connection is in the process of being refreshed."""

    CLOSING = "closing"
    """Connection is in the process of being closed."""

    CLOSED = "closed"
    """Connection is closed."""

INITIAL class-attribute instance-attribute

INITIAL = 'initial'

Connection is created and ready to open.

CONNECTING class-attribute instance-attribute

CONNECTING = 'connecting'

Connection is in the process of being established.

CONNECTED class-attribute instance-attribute

CONNECTED = 'connected'

Connection is established and operational.

REFRESHING class-attribute instance-attribute

REFRESHING = 'refreshing'

Connection is in the process of being refreshed.

CLOSING class-attribute instance-attribute

CLOSING = 'closing'

Connection is in the process of being closed.

CLOSED class-attribute instance-attribute

CLOSED = 'closed'

Connection is closed.

ConnectionProtocol

Bases: Protocol

Protocol describing a RabbitMQ connection interface.

Source code in rmqaio/rmqaio.py

class ConnectionProtocol(Protocol):
    """Protocol describing a RabbitMQ connection interface."""

    @property
    def url(self) -> str: ...

    @property
    def is_open(self) -> bool: ...

    @property
    def is_closed(self) -> bool: ...

    async def open(self, timeout: Number | None = None): ...

    async def refresh(self, timeout: Number | None = None): ...

    async def close(self, timeout: Number | None = None): ...

    async def new_channel(self, timeout: Number | None = None) -> aiormq.abc.AbstractChannel: ...

    async def channel(self, timeout: Number | None = None) -> aiormq.abc.AbstractChannel: ...

    def set_callback(
        self,
        name: str,
        callback: Callable[[ConnectionState, ConnectionState], Awaitable],
    ): ...

    async def remove_callback(self, name: str): ...

Connection

RabbitMQ connection with automatic reconnection on connection loss.

Attributes:

• id (str) –

  Connection Id.

• url (str) –

  Connection URL to RabbitMQ.

• ssl_context (SSLContext | None) –

  SSL context for TLS connections.

• open_retry_policy (RetryPolicy | None) –

  Policy for first connection attempts.

• reopen_retry_policy (RetryPolicy | None) –

  Policy for reconnection attempts.

• is_open (bool) –

  Whether connection is open and operational.

• is_closed (bool) –

  Whether connection is closed.

Examples:

>>> conn = Connection("amqp://localhost")
>>> await conn.open()
>>> channel = await conn.channel()
>>> await conn.close()

Source code in rmqaio/rmqaio.py

class Connection:
    """
    RabbitMQ connection with automatic reconnection on connection loss.

    Attributes:
        id: Connection Id.
        url: Connection URL to RabbitMQ.
        ssl_context: SSL context for TLS connections.
        open_retry_policy: Policy for first connection attempts.
        reopen_retry_policy: Policy for reconnection attempts.
        is_open: Whether connection is open and operational.
        is_closed: Whether connection is closed.

    Examples:
        >>> conn = Connection("amqp://localhost")
        >>> await conn.open()
        >>> channel = await conn.channel()
        >>> await conn.close()
    """

    def __init__(
        self,
        url: str,
        ssl_context: SSLContext | None = None,
        open_retry_policy: RetryPolicy | None = None,
        reopen_retry_policy: RetryPolicy | None = None,
    ):
        """
        Initialize connection.

        Args:
            url: AMQP connection URL (e.g., "amqp://localhost").
            ssl_context: Optional SSL context for TLS connections.
            open_retry_policy: Reconnection policy for handling first connection errors.
            reopen_retry_policy: Reconnection policy for handling reconnection errors.
        """
        self._id = uuid4().hex[-4:]
        self._url = url
        self._ssl_context = ssl_context
        self._open_retry_policy = open_retry_policy
        self._reopen_retry_policy = reopen_retry_policy or RetryPolicy()

        self._connect_timeout = self._extract_connect_timeout(url)
        self._state = ConnectionState.INITIAL
        self._conn: aiormq.Connection | None = None
        self._channel: aiormq.abc.AbstractChannel | None = None
        self._connected_event = asyncio.Event()
        self._closed_event = asyncio.Event()
        self._loop_task: asyncio.Task | None = None
        self._exc: BaseException | Exception | None = None
        self._callbacks: dict[str, Callable[[ConnectionState, ConnectionState], Awaitable]] = {}

    def _extract_connect_timeout(self, url: str) -> int | None:
        """
        Extract connection timeout from URL.

        Args:
            url: AMQP connection URL.

        Returns:
            Connection timeout in seconds, or None if not specified.
        """
        value = _as_int_or_none(yarl.URL(url).query.get("connection_timeout"))
        if value:
            value /= 1000  # milliseconds to seconds
        return cast(int, value)

    def __str__(self):
        url = yarl.URL(self._url)
        if url.port:
            return f"{self.__class__.__name__}[{url.host}:{url.port}][{self._id}]"
        return f"{self.__class__.__name__}[{url.host}][{self._id}]"

    def __repr__(self):
        return self.__str__()

    async def __aenter__(self):
        await self.open()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.close()

    @property
    def id(self) -> str:
        """Connection Id."""
        return self._id

    @property
    def url(self) -> str:
        """Connection URL."""
        return self._url

    @property
    def ssl_context(self) -> SSLContext | None:
        """SSL context for secure connections."""
        return self._ssl_context

    @property
    def open_retry_policy(self) -> RetryPolicy | None:
        """Reconnection policy for handling first connection errors."""
        return self._open_retry_policy

    @property
    def reopen_retry_policy(self) -> RetryPolicy | None:
        """Reconnection policy for handling reconnection errors."""
        return self._reopen_retry_policy

    @property
    def is_open(self) -> bool:
        """Whether connection is open and operational."""
        return self._state == ConnectionState.CONNECTED

    @property
    def is_closed(self) -> bool:
        """Whether connection is closed."""
        return self._state in [ConnectionState.CLOSED, ConnectionState.CLOSING]

    async def _start_loop(self):
        if self._state != ConnectionState.INITIAL:
            raise ConnectionInvalidStateError(_("invalid connection state"))
        if not self._loop_task:
            self._loop_task = asyncio.create_task(self._loop())

    async def _wait_open(self, timeout: Number | None = None):
        done = (
            await _wait_first_and_cancel_pending(
                [
                    create_task(self._connected_event.wait()),
                    create_task(self._closed_event.wait()),
                    create_task(wait([cast(asyncio.Task, self._loop_task)])),
                ],
                timeout=timeout,
            )
        )[0]
        if not done and self._loop_task:
            self._loop_task.cancel()
            with suppress(CancelledError):
                await self._loop_task

    def _check_open_result(self):
        if self._exc:
            raise self._exc
        if self._state == ConnectionState.CONNECTED:
            return
        raise ConnectionInvalidStateError(_("invalid connection state"))

    async def _start_refresh(self):
        await self._set_state(ConnectionState.REFRESHING)
        await cast(aiormq.Connection, self._conn).close()

    async def _wait_refresh(self, timeout: Number | None = None):
        await asyncio.wait_for(self._connected_event.wait(), timeout)

    async def _start_close(self):
        await self._set_state(ConnectionState.CLOSING)
        if self._loop_task:
            self._loop_task.cancel()

    async def _wait_close(self, timeout: Number | None = None):
        if self._loop_task:
            with suppress(CancelledError):
                await asyncio.wait_for(self._loop_task, timeout)

    async def open(self, timeout: Number | None = None):
        """
        Open connection to RabbitMQ.

        Establishes connection to RabbitMQ broker. If connection is lost,
        automatically attempts to reconnect based on retry policy.

        Args:
            timeout: Operation timeout in seconds.
        """
        if self._state == ConnectionState.CONNECTED:
            return

        if self._state in (ConnectionState.CONNECTING, ConnectionState.REFRESHING):
            await self._wait_open(timeout=timeout)
            self._check_open_result()
            return

        if self._state in [ConnectionState.CLOSING, ConnectionState.CLOSED]:
            raise ConnectionInvalidStateError("can not reopen closed connection")

        await self._start_loop()

        await self._wait_open(timeout=timeout)
        self._check_open_result()

    async def refresh(self, timeout: Number | None = None):
        """
        Refresh the underlying connection by reopening it.

        Args:
            timeout: Operation timeout in seconds.
        """
        if self._state in [ConnectionState.CLOSING, ConnectionState.CLOSED]:
            raise ConnectionInvalidStateError("can not refresh closed connection")

        if self._state != ConnectionState.CONNECTED:
            return

        await self._start_refresh()
        await self._wait_refresh(timeout=timeout)

        logger.info(_("%s refreshed"), self)

    async def close(self, timeout: Number | None = None):
        """
        Gracefully closes the connection and all associated channels.

        Args:
            timeout: Operation timeout in seconds.
        """
        if self._state in (ConnectionState.CLOSING, ConnectionState.CLOSED):
            return

        await self._start_close()
        await self._wait_close(timeout=timeout)

        logger.info(_("%s closed"), self)

    async def new_channel(self, timeout: Number | None = None) -> aiormq.abc.AbstractChannel:
        """
        Create a new channel.

        Args:
            timeout: Operation timeout in seconds.

        Returns:
            A new RabbitMQ channel.
        """
        if not self.is_open:
            await self.open()
        return await cast(aiormq.abc.AbstractConnection, self._conn).channel(timeout=timeout)

    async def channel(self, timeout: Number | None = None) -> aiormq.abc.AbstractChannel:
        """
        Get or create a channel.

        Args:
            timeout: Operation timeout in seconds.

        Returns:
            An open RabbitMQ channel.
        """
        if self._channel is None or self._channel.is_closed:
            await self.open()
            self._channel = await cast(aiormq.abc.AbstractConnection, self._conn).channel(timeout=timeout)
        return self._channel

    def set_callback(
        self,
        name: str,
        callback: Callable[[ConnectionState, ConnectionState], Awaitable],
    ):
        """
        Set a callback for connection events.

        Args:
            name: Unique name to identify the callback.
            callback: Callable to execute on event.

        If a callback with this name is already registered, it will be overridden.
        """
        self._callbacks[name] = callback

    async def remove_callback(self, name: str):
        """
        Remove a callback.

        Args:
            name: Callback name to remove.
        """
        self._callbacks.pop(name, None)

    async def _execute_callbacks(self, state_from: ConnectionState, state_to: ConnectionState):
        """
        Execute all registered callbacks.

        Args:
            state_from: Connection state before changing.
            state_to: Connection state after changing.
        """
        for name, callback in list(self._callbacks.items()):
            callback_logger.debug(
                _("%s execute callback[name=%s] %s -> %s"),
                self,
                name,
                state_from,
                state_to,
            )
            try:
                await callback(state_from, state_to)
            except Exception:
                callback_logger.exception(_("%s callback[name=%s, callback=%s] error"), self, name, callback)

    async def _set_state(self, state: ConnectionState):
        state_from, state_to, self._state = self._state, state, state
        if state_to != state_from:
            await self._execute_callbacks(state_from, state_to)

    async def _loop(self):
        """Main connection loop that manages connection lifecycle."""
        try:
            retry_policy = self._open_retry_policy
            delay_iter = iter(retry_policy.delays if retry_policy else [])

            while self._state not in [ConnectionState.CLOSING, ConnectionState.CLOSED]:
                try:
                    logger.info(_("%s connecting[timeout=%s]..."), self, self._connect_timeout)

                    await self._set_state(ConnectionState.CONNECTING)

                    self._conn = cast(
                        aiormq.Connection,
                        await wait_for(
                            aiormq.connect(self._url, context=self._ssl_context),
                            timeout=self._connect_timeout,
                        ),
                    )

                    self._channel = None
                    self._exc = None
                    self._connected_event.set()

                    logger.info(_("%s connected"), self)

                    await self._set_state(ConnectionState.CONNECTED)

                    retry_policy = self._reopen_retry_policy
                    delay_iter = iter(retry_policy.delays if retry_policy else [])

                    aws = [self._conn.closing, create_task(self._closed_event.wait())]
                    while True:
                        done = (await wait(aws, timeout=5, return_when=FIRST_COMPLETED))[0]
                        if done or (self._conn and self._conn.is_connection_was_stuck):
                            break

                    if not aws[1].done():
                        aws[1].cancel()
                        with suppress(CancelledError, RuntimeError):
                            await aws[1]

                    self._connected_event.clear()

                    if self._state in [ConnectionState.CLOSING, ConnectionState.CLOSED]:
                        break
                    elif self._state == ConnectionState.REFRESHING:
                        logger.warning(_("%s refreshing"), self)
                    else:
                        logger.warning(_("%s connection lost"), self)

                except asyncio.CancelledError as e:
                    self._exc = e

                    if self._conn and not self._conn.is_closed:
                        await self._conn.close()

                    raise e

                except Exception as e:
                    self._exc = e

                    if not retry_policy or not retry_policy.is_retryable(e):
                        break

                    try:
                        delay = next(delay_iter)
                    except StopIteration:
                        break

                    logger.warning(_("%s %s %s"), self, e.__class__, e)
                    logger.info(_("%s reconnecting in %.1f seconds"), self, delay)

                    await sleep(delay)

        except Exception as e:
            logger.exception(e)

        finally:
            with suppress(RuntimeError):
                self._connected_event.clear()

            self._conn = None
            self._channel = None
            self._loop_task = None

            if self._state == ConnectionState.CLOSING:
                self._closed_event.set()
                await self._set_state(ConnectionState.CLOSED)
            else:
                await self._set_state(ConnectionState.INITIAL)

id property

id: str

Connection Id.

url property

url: str

Connection URL.

ssl_context property

ssl_context: SSLContext | None

SSL context for secure connections.

open_retry_policy property

open_retry_policy: RetryPolicy | None

Reconnection policy for handling first connection errors.

reopen_retry_policy property

reopen_retry_policy: RetryPolicy | None

Reconnection policy for handling reconnection errors.

is_open property

is_open: bool

Whether connection is open and operational.

is_closed property

is_closed: bool

Whether connection is closed.

__init__

__init__(
    url: str,
    ssl_context: SSLContext | None = None,
    open_retry_policy: RetryPolicy | None = None,
    reopen_retry_policy: RetryPolicy | None = None,
)

Initialize connection.

Parameters:

• url (str) –

  AMQP connection URL (e.g., "amqp://localhost").

• ssl_context (SSLContext | None, default: None ) –

  Optional SSL context for TLS connections.

• open_retry_policy (RetryPolicy | None, default: None ) –

  Reconnection policy for handling first connection errors.

• reopen_retry_policy (RetryPolicy | None, default: None ) –

  Reconnection policy for handling reconnection errors.

Source code in rmqaio/rmqaio.py

def __init__(
    self,
    url: str,
    ssl_context: SSLContext | None = None,
    open_retry_policy: RetryPolicy | None = None,
    reopen_retry_policy: RetryPolicy | None = None,
):
    """
    Initialize connection.

    Args:
        url: AMQP connection URL (e.g., "amqp://localhost").
        ssl_context: Optional SSL context for TLS connections.
        open_retry_policy: Reconnection policy for handling first connection errors.
        reopen_retry_policy: Reconnection policy for handling reconnection errors.
    """
    self._id = uuid4().hex[-4:]
    self._url = url
    self._ssl_context = ssl_context
    self._open_retry_policy = open_retry_policy
    self._reopen_retry_policy = reopen_retry_policy or RetryPolicy()

    self._connect_timeout = self._extract_connect_timeout(url)
    self._state = ConnectionState.INITIAL
    self._conn: aiormq.Connection | None = None
    self._channel: aiormq.abc.AbstractChannel | None = None
    self._connected_event = asyncio.Event()
    self._closed_event = asyncio.Event()
    self._loop_task: asyncio.Task | None = None
    self._exc: BaseException | Exception | None = None
    self._callbacks: dict[str, Callable[[ConnectionState, ConnectionState], Awaitable]] = {}

open async

open(timeout: Number | None = None)

Open connection to RabbitMQ.

Establishes connection to RabbitMQ broker. If connection is lost, automatically attempts to reconnect based on retry policy.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

Source code in rmqaio/rmqaio.py

async def open(self, timeout: Number | None = None):
    """
    Open connection to RabbitMQ.

    Establishes connection to RabbitMQ broker. If connection is lost,
    automatically attempts to reconnect based on retry policy.

    Args:
        timeout: Operation timeout in seconds.
    """
    if self._state == ConnectionState.CONNECTED:
        return

    if self._state in (ConnectionState.CONNECTING, ConnectionState.REFRESHING):
        await self._wait_open(timeout=timeout)
        self._check_open_result()
        return

    if self._state in [ConnectionState.CLOSING, ConnectionState.CLOSED]:
        raise ConnectionInvalidStateError("can not reopen closed connection")

    await self._start_loop()

    await self._wait_open(timeout=timeout)
    self._check_open_result()

refresh async

refresh(timeout: Number | None = None)

Refresh the underlying connection by reopening it.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

Source code in rmqaio/rmqaio.py

async def refresh(self, timeout: Number | None = None):
    """
    Refresh the underlying connection by reopening it.

    Args:
        timeout: Operation timeout in seconds.
    """
    if self._state in [ConnectionState.CLOSING, ConnectionState.CLOSED]:
        raise ConnectionInvalidStateError("can not refresh closed connection")

    if self._state != ConnectionState.CONNECTED:
        return

    await self._start_refresh()
    await self._wait_refresh(timeout=timeout)

    logger.info(_("%s refreshed"), self)

close async

close(timeout: Number | None = None)

Gracefully closes the connection and all associated channels.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

Source code in rmqaio/rmqaio.py

async def close(self, timeout: Number | None = None):
    """
    Gracefully closes the connection and all associated channels.

    Args:
        timeout: Operation timeout in seconds.
    """
    if self._state in (ConnectionState.CLOSING, ConnectionState.CLOSED):
        return

    await self._start_close()
    await self._wait_close(timeout=timeout)

    logger.info(_("%s closed"), self)

new_channel async

new_channel(
    timeout: Number | None = None,
) -> aiormq.abc.AbstractChannel

Create a new channel.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

Returns:

• AbstractChannel –

  A new RabbitMQ channel.

Source code in rmqaio/rmqaio.py

async def new_channel(self, timeout: Number | None = None) -> aiormq.abc.AbstractChannel:
    """
    Create a new channel.

    Args:
        timeout: Operation timeout in seconds.

    Returns:
        A new RabbitMQ channel.
    """
    if not self.is_open:
        await self.open()
    return await cast(aiormq.abc.AbstractConnection, self._conn).channel(timeout=timeout)

channel async

channel(
    timeout: Number | None = None,
) -> aiormq.abc.AbstractChannel

Get or create a channel.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

Returns:

• AbstractChannel –

  An open RabbitMQ channel.

Source code in rmqaio/rmqaio.py

async def channel(self, timeout: Number | None = None) -> aiormq.abc.AbstractChannel:
    """
    Get or create a channel.

    Args:
        timeout: Operation timeout in seconds.

    Returns:
        An open RabbitMQ channel.
    """
    if self._channel is None or self._channel.is_closed:
        await self.open()
        self._channel = await cast(aiormq.abc.AbstractConnection, self._conn).channel(timeout=timeout)
    return self._channel

set_callback

set_callback(
    name: str,
    callback: Callable[
        [ConnectionState, ConnectionState], Awaitable
    ],
)

Set a callback for connection events.

Parameters:

• name (str) –

  Unique name to identify the callback.

• callback (Callable[[ConnectionState, ConnectionState], Awaitable]) –

  Callable to execute on event.

If a callback with this name is already registered, it will be overridden.

Source code in rmqaio/rmqaio.py

def set_callback(
    self,
    name: str,
    callback: Callable[[ConnectionState, ConnectionState], Awaitable],
):
    """
    Set a callback for connection events.

    Args:
        name: Unique name to identify the callback.
        callback: Callable to execute on event.

    If a callback with this name is already registered, it will be overridden.
    """
    self._callbacks[name] = callback

remove_callback async

remove_callback(name: str)

Remove a callback.

Parameters:

• name (str) –

  Callback name to remove.

Source code in rmqaio/rmqaio.py

async def remove_callback(self, name: str):
    """
    Remove a callback.

    Args:
        name: Callback name to remove.
    """
    self._callbacks.pop(name, None)

SharedConnection

Connection wrapper that shares a single underlying connection.

Instances created with identical parameters share the same Connection object. The underlying connection is closed only when all shared references are released.

Attributes:

• url (str) –

  Connection URL to RabbitMQ.

• ssl_context (SSLContext | None) –

  SSL context for TLS connections.

• open_retry_policy (RetryPolicy | None) –

  Policy for first connection attempts.

• reopen_retry_policy (RetryPolicy | None) –

  Policy for reconnection attempts.

• is_open (bool) –

  Whether connection is open and operational.

• is_closed (bool) –

  Whether connection is closed.

Examples:

>>> conn1 = SharedConnection("amqp://localhost")
>>> conn2 = SharedConnection("amqp://localhost")
>>> # conn1 and conn2 share the same underlying connection

Source code in rmqaio/rmqaio.py

class SharedConnection:
    """
    Connection wrapper that shares a single underlying connection.

    Instances created with identical parameters share the same `Connection` object.
    The underlying connection is closed only when all shared references are released.

    Attributes:
        url: Connection URL to RabbitMQ.
        ssl_context: SSL context for TLS connections.
        open_retry_policy: Policy for first connection attempts.
        reopen_retry_policy: Policy for reconnection attempts.
        is_open: Whether connection is open and operational.
        is_closed: Whether connection is closed.

    Examples:
        >>> conn1 = SharedConnection("amqp://localhost")
        >>> conn2 = SharedConnection("amqp://localhost")
        >>> # conn1 and conn2 share the same underlying connection
    """

    _shared = weakref.WeakValueDictionary()

    def __init__(
        self,
        url: str,
        ssl_context: SSLContext | None = None,
        open_retry_policy: RetryPolicy | None = None,
        reopen_retry_policy: RetryPolicy | None = None,
    ):
        """
        Initialize SharedConnection.

        Args:
            url: AMQP connection URL (e.g., "amqp://localhost").
            ssl_context: Optional SSL context for TLS connections.
            open_retry_policy: Reconnection policy for handling first connection errors.
            reopen_retry_policy: Reconnection policy for handling reconnection errors.
        """
        event_loop = get_running_loop()
        self._key = (
            event_loop,
            url,
            ssl_context,
            open_retry_policy,
            reopen_retry_policy,
        )
        if self._key not in self.__class__._shared:
            self._lock = Lock()
            self._conn = Connection(
                url,
                open_retry_policy=open_retry_policy,
                reopen_retry_policy=reopen_retry_policy,
                ssl_context=ssl_context,
            )
            self._shared_item = _SharedItem(lock=self._lock, conn=self._conn, refs=0)
            self.__class__._shared[self._key] = self._shared_item
        else:
            self._shared_item = self.__class__._shared[self._key]
            self._lock = self._shared_item.lock
            self._conn = self._shared_item.conn
        self._channel: aiormq.abc.AbstractChannel | None = None
        self._is_open = asyncio.Event()
        self._is_closed = asyncio.Event()

    def __str__(self):
        return f"{self.__class__.__name__}[{self._conn}]"

    def __repr__(self):
        return self.__str__()

    async def __aenter__(self):
        await self.open()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.close()

    @property
    def url(self) -> str:
        """Connection URL."""
        return self._conn.url

    @property
    def ssl_context(self) -> SSLContext | None:
        """SSL context for secure connections."""
        return self._conn.ssl_context

    @property
    def open_retry_policy(self) -> RetryPolicy | None:
        """Reconnection policy for handling first connection errors."""
        return self._conn._open_retry_policy

    @property
    def reopen_retry_policy(self) -> RetryPolicy | None:
        """Reconnection policy for handling reconnection errors."""
        return self._conn._reopen_retry_policy

    @property
    def is_open(self) -> bool:
        """Whether connection is open and operational."""
        return self._is_open.is_set() and self._conn.is_open

    @property
    def is_closed(self) -> bool:
        """Whether connection is closed."""
        return self._is_closed.is_set()

    async def open(self, timeout: Number | None = None):
        """
        Open connection to RabbitMQ.

        Establishes connection to RabbitMQ broker. If connection is lost,
        automatically attempts to reconnect based on retry policy.

        Args:
            timeout: Operation timeout in seconds.

        Raises:
            Exception: If connection fails, is closed, or reopened after close.
        """
        if self._is_closed.is_set():
            raise ConnectionInvalidStateError("can not reopen closed connection")

        async with self._lock:
            if self._is_open.is_set():
                return
            await self._conn.open(timeout=timeout)
            self._is_open.set()
            self._shared_item.refs += 1

    async def refresh(self, timeout: Number | None = None):
        """
        Refresh the underlying connection by reopening it.

        Args:
            timeout: Operation timeout in seconds.
        """
        async with self._lock:
            await self._conn.refresh(timeout=timeout)

    async def close(self, timeout: Number | None = None):
        """
        Gracefully closes the connection and all associated channels.

        Args:
            timeout: Operation timeout in seconds.
        """
        async with self._lock:
            if self._is_closed.is_set():
                return
            self._shared_item.refs = max(0, self._shared_item.refs - 1)
            if self._shared_item.refs == 0:
                await self._conn.close(timeout=timeout)
            self._is_closed.set()

    async def new_channel(self, timeout: Number | None = None) -> aiormq.abc.AbstractChannel:
        """
        Create a new channel.

        Args:
            timeout: Operation timeout in seconds.

        Returns:
            A new RabbitMQ channel.
        """
        await self.open()
        return await cast(Connection, self._conn).new_channel(timeout=timeout)

    async def channel(self, timeout: Number | None = None) -> aiormq.abc.AbstractChannel:
        """
        Get or create a channel.

        Args:
            timeout: Operation timeout in seconds.

        Returns:
            An open RabbitMQ channel.
        """
        if self._channel is None or self._channel.is_closed:
            await self.open()
            self._channel = await self._conn.new_channel(timeout=timeout)
        return self._channel

    def set_callback(
        self,
        name: str,
        callback: Callable[[ConnectionState, ConnectionState], Awaitable],
    ):
        """
        Set a callback for connection events.

        Args:
            name: Unique name to identify the callback.
            callback: Callable to execute on event.

        If a callback with this name is already registered, it will be overridden.
        """
        return self._conn.set_callback(name, callback)

    async def remove_callback(self, name: str):
        """
        Remove a callback.

        Args:
            name: Callback name to remove.
        """
        return await self._conn.remove_callback(name)

url property

url: str

Connection URL.

ssl_context property

ssl_context: SSLContext | None

SSL context for secure connections.

open_retry_policy property

open_retry_policy: RetryPolicy | None

Reconnection policy for handling first connection errors.

reopen_retry_policy property

reopen_retry_policy: RetryPolicy | None

Reconnection policy for handling reconnection errors.

is_open property

is_open: bool

Whether connection is open and operational.

is_closed property

is_closed: bool

Whether connection is closed.

__init__

__init__(
    url: str,
    ssl_context: SSLContext | None = None,
    open_retry_policy: RetryPolicy | None = None,
    reopen_retry_policy: RetryPolicy | None = None,
)

Initialize SharedConnection.

Parameters:

• url (str) –

  AMQP connection URL (e.g., "amqp://localhost").

• ssl_context (SSLContext | None, default: None ) –

  Optional SSL context for TLS connections.

• open_retry_policy (RetryPolicy | None, default: None ) –

  Reconnection policy for handling first connection errors.

• reopen_retry_policy (RetryPolicy | None, default: None ) –

  Reconnection policy for handling reconnection errors.

Source code in rmqaio/rmqaio.py

def __init__(
    self,
    url: str,
    ssl_context: SSLContext | None = None,
    open_retry_policy: RetryPolicy | None = None,
    reopen_retry_policy: RetryPolicy | None = None,
):
    """
    Initialize SharedConnection.

    Args:
        url: AMQP connection URL (e.g., "amqp://localhost").
        ssl_context: Optional SSL context for TLS connections.
        open_retry_policy: Reconnection policy for handling first connection errors.
        reopen_retry_policy: Reconnection policy for handling reconnection errors.
    """
    event_loop = get_running_loop()
    self._key = (
        event_loop,
        url,
        ssl_context,
        open_retry_policy,
        reopen_retry_policy,
    )
    if self._key not in self.__class__._shared:
        self._lock = Lock()
        self._conn = Connection(
            url,
            open_retry_policy=open_retry_policy,
            reopen_retry_policy=reopen_retry_policy,
            ssl_context=ssl_context,
        )
        self._shared_item = _SharedItem(lock=self._lock, conn=self._conn, refs=0)
        self.__class__._shared[self._key] = self._shared_item
    else:
        self._shared_item = self.__class__._shared[self._key]
        self._lock = self._shared_item.lock
        self._conn = self._shared_item.conn
    self._channel: aiormq.abc.AbstractChannel | None = None
    self._is_open = asyncio.Event()
    self._is_closed = asyncio.Event()

open async

open(timeout: Number | None = None)

Open connection to RabbitMQ.

Establishes connection to RabbitMQ broker. If connection is lost, automatically attempts to reconnect based on retry policy.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

Raises:

• Exception –

  If connection fails, is closed, or reopened after close.

Source code in rmqaio/rmqaio.py

async def open(self, timeout: Number | None = None):
    """
    Open connection to RabbitMQ.

    Establishes connection to RabbitMQ broker. If connection is lost,
    automatically attempts to reconnect based on retry policy.

    Args:
        timeout: Operation timeout in seconds.

    Raises:
        Exception: If connection fails, is closed, or reopened after close.
    """
    if self._is_closed.is_set():
        raise ConnectionInvalidStateError("can not reopen closed connection")

    async with self._lock:
        if self._is_open.is_set():
            return
        await self._conn.open(timeout=timeout)
        self._is_open.set()
        self._shared_item.refs += 1

refresh async

refresh(timeout: Number | None = None)

Refresh the underlying connection by reopening it.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

Source code in rmqaio/rmqaio.py

async def refresh(self, timeout: Number | None = None):
    """
    Refresh the underlying connection by reopening it.

    Args:
        timeout: Operation timeout in seconds.
    """
    async with self._lock:
        await self._conn.refresh(timeout=timeout)

close async

close(timeout: Number | None = None)

Gracefully closes the connection and all associated channels.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

Source code in rmqaio/rmqaio.py

async def close(self, timeout: Number | None = None):
    """
    Gracefully closes the connection and all associated channels.

    Args:
        timeout: Operation timeout in seconds.
    """
    async with self._lock:
        if self._is_closed.is_set():
            return
        self._shared_item.refs = max(0, self._shared_item.refs - 1)
        if self._shared_item.refs == 0:
            await self._conn.close(timeout=timeout)
        self._is_closed.set()

new_channel async

new_channel(
    timeout: Number | None = None,
) -> aiormq.abc.AbstractChannel

Create a new channel.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

Returns:

• AbstractChannel –

  A new RabbitMQ channel.

Source code in rmqaio/rmqaio.py

async def new_channel(self, timeout: Number | None = None) -> aiormq.abc.AbstractChannel:
    """
    Create a new channel.

    Args:
        timeout: Operation timeout in seconds.

    Returns:
        A new RabbitMQ channel.
    """
    await self.open()
    return await cast(Connection, self._conn).new_channel(timeout=timeout)

channel async

channel(
    timeout: Number | None = None,
) -> aiormq.abc.AbstractChannel

Get or create a channel.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

Returns:

• AbstractChannel –

  An open RabbitMQ channel.

Source code in rmqaio/rmqaio.py

async def channel(self, timeout: Number | None = None) -> aiormq.abc.AbstractChannel:
    """
    Get or create a channel.

    Args:
        timeout: Operation timeout in seconds.

    Returns:
        An open RabbitMQ channel.
    """
    if self._channel is None or self._channel.is_closed:
        await self.open()
        self._channel = await self._conn.new_channel(timeout=timeout)
    return self._channel

set_callback

set_callback(
    name: str,
    callback: Callable[
        [ConnectionState, ConnectionState], Awaitable
    ],
)

Set a callback for connection events.

Parameters:

• name (str) –

  Unique name to identify the callback.

• callback (Callable[[ConnectionState, ConnectionState], Awaitable]) –

  Callable to execute on event.

If a callback with this name is already registered, it will be overridden.

Source code in rmqaio/rmqaio.py

def set_callback(
    self,
    name: str,
    callback: Callable[[ConnectionState, ConnectionState], Awaitable],
):
    """
    Set a callback for connection events.

    Args:
        name: Unique name to identify the callback.
        callback: Callable to execute on event.

    If a callback with this name is already registered, it will be overridden.
    """
    return self._conn.set_callback(name, callback)

remove_callback async

remove_callback(name: str)

Remove a callback.

Parameters:

• name (str) –

  Callback name to remove.

Source code in rmqaio/rmqaio.py

async def remove_callback(self, name: str):
    """
    Remove a callback.

    Args:
        name: Callback name to remove.
    """
    return await self._conn.remove_callback(name)

DefaultExchangeSpec dataclass

Default exchange specification.

Attributes:

• name (Literal['']) –

  Always empty string.

• kind (Literal['default']) –

  Always "default".

• type (ExchangeType) –

  Exchange type.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class DefaultExchangeSpec:
    """
    Default exchange specification.

    Attributes:
        name: Always empty string.
        kind: Always "default".
        type: Exchange type.
    """

    name: Literal[""] = field(init=False, default="")
    kind: Literal["default"] = field(init=False, default="default")
    type: ExchangeType = "direct"

BaseExchangeArgs dataclass

Base exchange arguments.

Attributes:

• alternate_exchange (str | None) –

  Alternate exchange name.

• internal (bool | None) –

  Whether exchange is internal.

• custom (dict[str, Any] | None) –

  Custom arguments.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class BaseExchangeArgs:
    """
    Base exchange arguments.

    Attributes:
        alternate_exchange: Alternate exchange name.
        internal: Whether exchange is internal.
        custom: Custom arguments.
    """

    alternate_exchange: str | None = None
    internal: bool | None = None
    custom: dict[str, Any] | None = None

    def to_dict(self) -> dict[str, Any]:
        args: dict[str, Any] = {}

        if self.alternate_exchange is not None:
            args["alternate-exchange"] = self.alternate_exchange

        if self.internal is not None:
            args["internal"] = self.internal

        if self.custom is not None:
            args.update(self.custom)

        return args

BaseExchangeSpec dataclass

Base exchange specification.

Attributes:

• name (str) –

  Exchange name.

• kind (str) –

  Exchange kind ("normal" or "read-only").

• type (str) –

  Exchange type.

• durable (bool) –

  Whether exchange is durable.

• auto_delete (bool) –

  Whether to delete when no longer used.

• internal (bool) –

  Whether exchange is internal.

• arguments (BaseExchangeArgs) –

  Exchange arguments.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class BaseExchangeSpec:
    """
    Base exchange specification.

    Attributes:
        name: Exchange name.
        kind: Exchange kind ("normal" or "read-only").
        type: Exchange type.
        durable: Whether exchange is durable.
        auto_delete: Whether to delete when no longer used.
        internal: Whether exchange is internal.
        arguments: Exchange arguments.
    """

    name: str
    kind: str
    type: str
    durable: bool = True
    auto_delete: bool = False
    internal: bool = False

    arguments: BaseExchangeArgs = field(default_factory=BaseExchangeArgs)

    def __hash__(self):
        return hash(self.name)

ExchangeArgs dataclass

Bases: BaseExchangeArgs

Exchange arguments.

Inherits all attributes from BaseExchangeArgs.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class ExchangeArgs(BaseExchangeArgs):
    """
    Exchange arguments.

    Inherits all attributes from BaseExchangeArgs.
    """

ExchangeSpec dataclass

Bases: BaseExchangeSpec

Exchange specification.

Attributes:

• kind (Literal['normal', 'read-only']) –

  Always "normal" or "read-only".

• type (ExchangeType) –

  Exchange type.

• arguments (ExchangeArgs) –

  Exchange arguments.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class ExchangeSpec(BaseExchangeSpec):
    """
    Exchange specification.

    Attributes:
        kind: Always "normal" or "read-only".
        type: Exchange type.
        arguments: Exchange arguments.
    """

    kind: Literal["normal", "read-only"] = field(default="normal")
    type: ExchangeType = "direct"

    arguments: ExchangeArgs = field(default_factory=ExchangeArgs)

    def __post_init__(self):
        if self.name == "":
            raise ValueError("use DefaultExchangeSpec for default exchange instead of name=''")

BaseQueueSpec dataclass

Base queue specification.

Attributes:

• name (str) –

  Queue name.

• kind (str) –

  Queue kind ("normal" or "read-only").

• durable (bool) –

  Whether queue is durable.

• exclusive (bool) –

  Whether queue is exclusive.

• auto_delete (bool) –

  Whether to delete when no longer used.

• arguments (BaseQueueArgs) –

  Queue arguments.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class BaseQueueSpec:
    """
    Base queue specification.

    Attributes:
        name: Queue name.
        kind: Queue kind ("normal" or "read-only").
        durable: Whether queue is durable.
        exclusive: Whether queue is exclusive.
        auto_delete: Whether to delete when no longer used.
        arguments: Queue arguments.
    """

    name: str
    kind: str

    durable: bool = True
    exclusive: bool = False
    auto_delete: bool = False

    arguments: BaseQueueArgs = field(default_factory=BaseQueueArgs)

    def __hash__(self):
        return hash(self.name)

QueueArgs dataclass

Bases: BaseQueueArgs

Queue arguments.

Attributes:

• queue_type (QueueType) –

  Queue type (defaults to "classic").

Inherits all attributes from BaseQueueArgs.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class QueueArgs(BaseQueueArgs):
    """
    Queue arguments.

    Attributes:
        queue_type: Queue type (defaults to "classic").

    Inherits all attributes from BaseQueueArgs.
    """

    queue_type: QueueType = "classic"

QueueSpec dataclass

Bases: BaseQueueSpec

Queue specification.

Attributes:

• kind (Literal['normal', 'read-only']) –

  Always "normal" or "read-only".

• arguments (QueueArgs) –

  Queue arguments.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class QueueSpec(BaseQueueSpec):
    """
    Queue specification.

    Attributes:
        kind: Always "normal" or "read-only".
        arguments: Queue arguments.
    """

    kind: Literal["normal", "read-only"] = field(default="normal")

    arguments: QueueArgs = field(default_factory=QueueArgs)

BindSpec dataclass

Binding specification.

Attributes:

• src (str) –

  Source exchange or queue.

• dst (str) –

  Destination exchange or queue.

• routing_key (str) –

  Routing key.

• kind (Literal['exchange', 'queue']) –

  Type of binding ("exchange" or "queue").

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class BindSpec:
    """
    Binding specification.

    Attributes:
        src: Source exchange or queue.
        dst: Destination exchange or queue.
        routing_key: Routing key.
        kind: Type of binding ("exchange" or "queue").
    """

    src: str
    dst: str
    routing_key: str
    kind: Literal["exchange", "queue"] = "queue"

ConsumerArgs dataclass

Consumer arguments.

Attributes:

• priority (int | None) –

  Consumer priority.

• cancel_on_ha_failover (bool | None) –

  Cancel on HA failover.

• custom (dict[str, Any] | None) –

  Custom arguments.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class ConsumerArgs:
    """
    Consumer arguments.

    Attributes:
        priority: Consumer priority.
        cancel_on_ha_failover: Cancel on HA failover.
        custom: Custom arguments.
    """

    priority: int | None = None
    cancel_on_ha_failover: bool | None = None

    custom: dict[str, Any] | None = None

    def to_dict(self) -> dict[str, Any]:
        args: dict[str, Any] = {}

        if self.priority is not None:
            args["x-priority"] = self.priority

        if self.cancel_on_ha_failover is not None:
            args["x-cancel-on-ha-failover"] = self.cancel_on_ha_failover

        if self.custom is not None:
            args.update(self.custom)

        return args

ConsumerSpec dataclass

Consumer specification.

Attributes:

• queue (str) –

  Queue name.

• callback (Callable[[AbstractChannel, DeliveredMessage], Awaitable]) –

  Async callback to process messages.

• prefetch_count (int | None) –

  Maximum unacknowledged messages.

• prefetch_size (int | None) –

  Maximum unacknowledged bytes.

• auto_ack (bool) –

  Auto acknowledge messages.

• exclusive (bool) –

  Exclusive consumer.

• arguments (ConsumerArgs) –

  Consumer arguments.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class ConsumerSpec:
    """
    Consumer specification.

    Attributes:
        queue: Queue name.
        callback: Async callback to process messages.
        prefetch_count: Maximum unacknowledged messages.
        prefetch_size: Maximum unacknowledged bytes.
        auto_ack: Auto acknowledge messages.
        exclusive: Exclusive consumer.
        arguments: Consumer arguments.
    """

    queue: str
    callback: Callable[[aiormq.abc.AbstractChannel, aiormq.abc.DeliveredMessage], Awaitable]
    prefetch_count: int | None = None
    prefetch_size: int | None = None
    auto_ack: bool = True
    exclusive: bool = False
    consumer_tag: str | None = None

    arguments: ConsumerArgs = field(default_factory=ConsumerArgs)

Consumer dataclass

Consumer instance.

Attributes:

• spec (ConsumerSpec) –

  Consumer specification.

• consumer_tag (str) –

  Consumer tag from broker.

• channel (AbstractChannel) –

  AMQP channel for this consumer.

Source code in rmqaio/rmqaio.py

@dataclass(slots=True, frozen=True)
class Consumer:
    """
    Consumer instance.

    Attributes:
        spec: Consumer specification.
        consumer_tag: Consumer tag from broker.
        channel: AMQP channel for this consumer.
    """

    spec: ConsumerSpec
    consumer_tag: str
    channel: aiormq.abc.AbstractChannel

Topology dataclass

Topology of RabbitMQ entities.

Attributes:

• exchanges (UniqueList[BaseExchangeSpec]) –

  List of exchange specifications.

• queues (UniqueList[BaseQueueSpec]) –

  List of queue specifications.

• bindings (UniqueList[BindSpec]) –

  List of binding specifications.

• consumers (UniqueList[ConsumerSpec]) –

  List of consumer specifications.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class Topology:
    """
    Topology of RabbitMQ entities.

    Attributes:
        exchanges: List of exchange specifications.
        queues: List of queue specifications.
        bindings: List of binding specifications.
        consumers: List of consumer specifications.
    """

    exchanges: UniqueList[BaseExchangeSpec] = field(default_factory=lambda: UniqueList[BaseExchangeSpec]())
    queues: UniqueList[BaseQueueSpec] = field(default_factory=lambda: UniqueList[BaseQueueSpec]())
    bindings: UniqueList[BindSpec] = field(default_factory=lambda: UniqueList[BindSpec]())
    consumers: UniqueList[ConsumerSpec] = field(default_factory=lambda: UniqueList[ConsumerSpec]())

Ops

RabbitMQ operations handler.

Provides high-level operations for managing exchanges, queues, bindings, and consuming messages.

Attributes:

• conn –

  Connection to RabbitMQ.

• timeout –

  Default operation timeout.

Source code in rmqaio/rmqaio.py

class Ops:
    """
    RabbitMQ operations handler.

    Provides high-level operations for managing exchanges, queues,
    bindings, and consuming messages.

    Attributes:
        conn: Connection to RabbitMQ.
        timeout: Default operation timeout.
    """

    def __init__(
        self,
        conn: ConnectionProtocol,
        timeout: Number | None = None,
    ):
        """
        Initialize Ops.

        Args:
            conn: Connection to RabbitMQ.
            timeout: Default operation timeout.
        """
        self._conn = conn
        self._timeout = timeout
        self._topology = Topology()
        self._consumers: dict[str, Consumer] = {}

        self._conn.set_callback(
            f"on_connection_state_changed[{id(self)}]",
            self._on_connection_state_changed,
        )

    async def _on_connection_state_changed(self, state_from: ConnectionState, state_to: ConnectionState):
        if state_to == ConnectionState.CONNECTED and state_from != ConnectionState.INITIAL:
            await self._restore_topology()

    async def _restore_topology(self):
        for spec in self._topology.exchanges:
            await self.exchange_declare(spec)
        for spec in self._topology.queues:
            await self.queue_declare(spec)
        for spec in self._topology.bindings:
            await self.bind(spec)

        consumers_map = {consumer.spec: consumer for consumer in self._consumers.values()}
        for spec in self._topology.consumers:
            consumer = consumers_map.get(spec)
            if consumer and not consumer.channel.is_closed:
                continue
            await self.consume(spec)

    @property
    def consumers(self) -> list[Consumer]:
        return list(self._consumers.values())

    async def apply_topology(
        self,
        topology: Topology,
        consume: bool | None = None,
        restore: bool | None = None,
        force: bool | None = None,
    ):
        """
        Apply entire topology declaration.

        Args:
            topology: Topology to apply.
            consume: If `True`, start consuming according to the topology.
            restore: If `True`, restore on reconnect.
        """
        logger.info("apply topology[restore=%s] %s", topology, restore)

        for spec in topology.exchanges:
            await self.exchange_declare(spec, restore=restore, force=force)
        for spec in topology.queues:
            await self.queue_declare(spec, restore=restore, force=force)
        for spec in topology.bindings:
            await self.bind(spec, restore=restore)
        if consume:
            for spec in topology.consumers:
                if not next(filter(lambda consumer: consumer.spec == spec, self.consumers), None):
                    await self.consume(spec, restore=restore)

    async def check_exists(
        self,
        spec: BaseExchangeSpec | BaseQueueSpec,
        timeout: Number | None = None,
    ) -> bool:
        """
        Check if subj exists.

        Args:
            spec: Specification to check.
            timeout: Operation timeout in seconds. If `None`, uses the default timeout.

        Returns:
            True if the subj exists, False otherwise.
        """
        match spec:
            case BaseExchangeSpec():
                return await self.check_exchange_exists(spec.name, timeout=timeout)
            case BaseQueueSpec():
                return await self.check_queue_exists(spec.name, timeout=timeout)
            case _:
                raise ValueError("invalid spec type")

    async def declare(
        self,
        spec: BaseExchangeSpec | BaseQueueSpec,
        timeout: Number | None = None,
        restore: bool | None = None,
        force: bool | None = None,
    ):
        """
        Declare subj.

        Args:
            restore: If `True`, automatically redeclare subj after reconnection.
            force: If `True`, delete and redeclare subj if declaration fails
                due to parameter mismatch.
            timeout: Operation timeout. If `None`, uses the default timeout.
        """
        match spec:
            case BaseExchangeSpec():
                await self.exchange_declare(spec, timeout=timeout, restore=restore, force=force)
            case BaseQueueSpec():
                await self.queue_declare(spec, timeout=timeout, restore=restore, force=force)
            case _:
                raise ValueError("invalid spec type")

    async def delete(self, spec: Spec, timeout: Number | None = None):
        """
        Delete subj.

        Args:
            timeout: Operation timeout. If `None`, uses the default timeout.

        Raises:
            Exception: If deleting fails.
        """
        match spec:
            case BaseExchangeSpec():
                if spec.kind == "read-only":
                    raise OperationError("can not delete read-only exchange")
                await self.exchange_delete(spec.name, timeout=timeout)
            case BaseQueueSpec():
                if spec.kind == "read-only":
                    raise OperationError("can not delete read-only queue")
                await self.queue_delete(spec.name, timeout=timeout)
            case _:
                raise ValueError("invalid spec type")

    async def check_exchange_exists(self, name: str, timeout: Number | None = None) -> bool:
        """
        Check if exchange exists.

        Args:
            name: Exchange name.
            timeout: Operation timeout in seconds. If `None`, uses the default timeout.

        Returns:
            True if the exchange exists, False otherwise.
        """
        timeout_ = timeout if timeout is not None else self._timeout

        channel = await self._conn.channel(timeout=timeout_)

        try:
            await channel.exchange_declare(name, passive=True, timeout=timeout_)
            return True
        except aiormq.ChannelNotFoundEntity:
            return False

    async def exchange_declare(
        self,
        spec: BaseExchangeSpec,
        timeout: Number | None = None,
        restore: bool | None = None,
        force: bool | None = None,
    ):
        """
        Declare the exchange.

        Args:
            timeout: Operation timeout in seconds. If `None`, uses the default timeout.
            restore: If `True`, automatically redeclare the exchange after reconnection.
            force: If `True`, delete and redeclare the exchange if declaration fails
                due to parameter mismatch.

        Raises:
            Exception: If declaring fails.
        """
        if spec.kind == "read-only":
            raise OperationError("can not declare read-only exchange")

        timeout_ = timeout if timeout is not None else self._timeout

        async def op():
            logger.info(_("declare[restore=%s, force=%s] %s"), restore, force, spec)

            channel = await self._conn.channel(timeout=timeout_)
            await channel.exchange_declare(
                spec.name,
                exchange_type=spec.type,
                durable=spec.durable,
                auto_delete=spec.auto_delete,
                arguments=spec.arguments.to_dict(),
                timeout=timeout_,
            )

        if force:

            async def on_error(e):
                channel = await self._conn.channel(timeout=timeout_)
                logger.info(_("delete[on_error] %s"), spec)
                await channel.exchange_delete(spec.name, timeout=timeout_)

            await retry(
                RetryPolicy(delays=[0], exc_filter=(aiormq.ChannelPreconditionFailed,)),
                on_error=on_error,
            )(op)()

        else:
            await op()

        if restore:
            self._topology.exchanges.append(spec)

    async def exchange_delete(self, name: str, timeout: Number | None = None):
        """
        Delete exchange.

        Args:
            name: Exchange name.
            timeout: Operation timeout. If `None`, uses the default timeout.
        """
        logger.info(_("delete exchange '%s'"), name)

        timeout_ = timeout if timeout is not None else self._timeout

        channel = await self._conn.channel(timeout=timeout_)

        await channel.exchange_delete(name, timeout=timeout_)

        spec = next(filter(lambda spec: spec.name == name, self._topology.exchanges), None)
        if spec:
            self._topology.exchanges.remove(spec)

    async def check_queue_exists(self, name: str, timeout: Number | None = None) -> bool:
        """
        Check if queue exists.

        Args:
            name: Queue name.
            timeout: Operation timeout. If `None`, uses the default timeout.

        Returns:
            True if the queue exists, False otherwise.
        """
        timeout_ = timeout if timeout is not None else self._timeout

        channel = await self._conn.channel(timeout=timeout_)

        try:
            await channel.queue_declare(name, passive=True, timeout=timeout_)
            return True
        except aiormq.ChannelNotFoundEntity:
            return False

    async def queue_declare(
        self,
        spec: BaseQueueSpec,
        timeout: Number | None = None,
        restore: bool | None = None,
        force: bool | None = None,
    ):
        """
        Declare queue.

        Args:
            spec: Queue specification.
            timeout: Operation timeout. If `None`, uses the default timeout.
            restore: Restore this binding on connection issue.
            force: Force redeclare queue if it has already been declared with different parameters.

        Raises:
            Exception: If declaring fails.
        """
        if spec.kind == "read-only":
            raise OperationError("can not declare read-only queue")

        timeout_ = timeout if timeout is not None else self._timeout

        async def op():
            logger.info(_("declare[restore=%s, force=%s] %s"), restore, force, spec)

            channel = await self._conn.channel(timeout=timeout_)
            arguments = spec.arguments.to_dict()
            await channel.queue_declare(
                spec.name,
                durable=spec.durable,
                exclusive=spec.exclusive,
                auto_delete=spec.auto_delete,
                arguments=arguments,
                timeout=timeout_,
            )

        if force:

            async def on_error(e):
                channel = await self._conn.channel(timeout=timeout_)
                logger.info(_("delete[on_error] %s"), spec)
                await channel.queue_delete(spec.name, timeout=timeout_)

            await retry(
                RetryPolicy(delays=[0], exc_filter=(aiormq.ChannelPreconditionFailed,)),
                on_error=on_error,
            )(op)()

        else:
            await op()

        if restore:
            self._topology.queues.append(spec)

    async def queue_delete(self, name: str, timeout: Number | None = None):
        """
        Delete queue.

        Args:
            name: Queue name.
            timeout: Operation timeout. If `None`, uses the default timeout.
        """
        logger.info(_("delete queue '%s'"), name)

        timeout_ = timeout if timeout is not None else self._timeout

        channel = await self._conn.channel(timeout=timeout_)

        await channel.queue_delete(name, timeout=timeout_)

        spec = next(filter(lambda spec: spec.name == name, self._topology.queues), None)
        if spec:
            self._topology.queues.remove(spec)

    async def bind(
        self,
        spec: BindSpec,
        timeout: Number | None = None,
        restore: bool | None = None,
    ):
        """
        Bind item to exchange.

        Args:
            spec: Bind specification.
            timeout: Operation timeout. If `None`, uses the default timeout.
            restore: Restore this binding on connection issue.
        """
        logger.info(
            _("bind %s '%s' to exchange '%s' with routing_key '%s'"),
            spec.kind,
            spec.dst,
            spec.src,
            spec.routing_key,
        )

        timeout_ = timeout if timeout is not None else self._timeout
        channel = await self._conn.channel(timeout=timeout_)

        match spec.kind:
            case "exchange":
                await channel.exchange_bind(
                    spec.dst,
                    spec.src,
                    routing_key=spec.routing_key,
                    timeout=timeout_,
                )
            case "queue":
                await channel.queue_bind(
                    spec.dst,
                    spec.src,
                    routing_key=spec.routing_key,
                    timeout=timeout_,
                )
            case _:
                raise ValueError("invalid spec type")

        if restore:
            self._topology.bindings.append(spec)

    async def unbind(
        self,
        spec: BindSpec,
        timeout: Number | None = None,
    ):
        """
        Unbind item from exchange.

        Args:
            spec: Bind specification.
            timeout: Operation timeout. If `None`, uses the default timeout.
        """
        logger.info(
            _("unbind %s '%s' from exchange '%s' for routing_key '%s'"),
            spec.kind,
            spec.dst,
            spec.src,
            spec.routing_key,
        )

        timeout_ = timeout if timeout is not None else self._timeout
        channel = await self._conn.channel(timeout=timeout_)

        match spec.kind:
            case "exchange":
                await channel.exchange_unbind(
                    spec.dst,
                    spec.src,
                    routing_key=spec.routing_key,
                    timeout=timeout_,
                )
            case "queue":
                await channel.queue_unbind(
                    spec.dst,
                    spec.src,
                    routing_key=spec.routing_key,
                    timeout=timeout_,
                )
            case _:
                raise ValueError("invalid spec type")

        if spec in self._topology.bindings:
            self._topology.bindings.remove(spec)

    async def publish(
        self,
        exchange: str,
        data: bytes,
        routing_key: str,
        properties: dict | None = None,
        mandatory: bool = False,
        timeout: Number | None = None,
    ):
        """
        Publish data to the exchange.

        Args:
            exchange: Exchange name.
            data: Data to publish.
            routing_key: Routing key for message delivery.
            properties: Optional RabbitMQ message properties.
            mandatory: If `True`, return unroutable message to publisher.
            timeout: Operation timeout in seconds. If `None`, uses the default timeout.
        """
        timeout_ = timeout if timeout is not None else self._timeout

        channel = await self._conn.channel(timeout=timeout_)

        if logger.isEnabledFor(logging.DEBUG):
            logger.debug(
                _("exchange[name='%s'] %s channel[%s] publish[routing_key='%s'] %s"),
                exchange,
                self._conn,
                channel,
                routing_key,
                (
                    (
                        data[: config.log_data_truncate_size] + b"<truncated>"
                        if len(data) > config.log_data_truncate_size
                        else data
                    )
                    if not config.log_sanitize
                    else "<hidden>"
                ),
            )

        await channel.basic_publish(
            data,
            exchange=exchange,
            routing_key=routing_key,
            properties=BasicProperties(**(properties or {})),
            mandatory=mandatory,
            timeout=timeout_,
        )

    async def consume(
        self,
        spec: ConsumerSpec,
        timeout: Number | None = None,
        restore: bool | None = None,
    ) -> Consumer:
        """
        Consume queue.

        Args:
            spec: Spec.
            timeout: Operation timeout. If `None`, uses the default timeout.
            restore: Restore consuming on connection issue.

        Returns:
            Consumer: The active consumer instance.
        """
        timeout_ = timeout if timeout is not None else self._timeout

        channel = await self._conn.new_channel(timeout=timeout_)

        await channel.basic_qos(
            prefetch_count=spec.prefetch_count,
            prefetch_size=spec.prefetch_size,
            timeout=timeout_,
        )

        consumer_tag = cast(
            str,
            (
                await channel.basic_consume(
                    spec.queue,
                    partial(spec.callback, channel),
                    no_ack=spec.auto_ack,
                    exclusive=spec.exclusive,
                    arguments=spec.arguments.to_dict(),
                    consumer_tag=spec.consumer_tag,
                    timeout=timeout_,
                )
            ).consumer_tag,
        )

        logger.info(_("consume %s"), spec)

        consumer = Consumer(spec, consumer_tag, channel)

        if restore:
            self._topology.consumers.append(spec)

        self._consumers[consumer_tag] = consumer

        return consumer

    async def stop_consume(
        self,
        consumer_tag: str | None = None,
        timeout: Number | None = None,
    ):
        """
        Stop consume.

        Args:
            consumer_tag: Consumer tag. If `None`, stop all consumers.
            timeout: Operation timeout. If `None`, uses the default timeout.
        """
        if consumer_tag:
            tags = [consumer_tag]
        else:
            tags = list(self._consumers.keys())
        for tag in tags:
            if tag in self._consumers:
                consumer = self._consumers.pop(tag)
                if consumer.spec in self._topology.consumers:
                    self._topology.consumers.remove(consumer.spec)
                if not consumer.channel.is_closed:
                    logger.info(_("stop consume %s"), consumer.spec)
                    timeout_ = timeout if timeout is not None else self._timeout
                    await consumer.channel.basic_cancel(consumer.consumer_tag, timeout=timeout_)

__init__

__init__(
    conn: ConnectionProtocol, timeout: Number | None = None
)

Initialize Ops.

Parameters:

• conn (ConnectionProtocol) –

  Connection to RabbitMQ.

• timeout (Number | None, default: None ) –

  Default operation timeout.

Source code in rmqaio/rmqaio.py

def __init__(
    self,
    conn: ConnectionProtocol,
    timeout: Number | None = None,
):
    """
    Initialize Ops.

    Args:
        conn: Connection to RabbitMQ.
        timeout: Default operation timeout.
    """
    self._conn = conn
    self._timeout = timeout
    self._topology = Topology()
    self._consumers: dict[str, Consumer] = {}

    self._conn.set_callback(
        f"on_connection_state_changed[{id(self)}]",
        self._on_connection_state_changed,
    )

apply_topology async

apply_topology(
    topology: Topology,
    consume: bool | None = None,
    restore: bool | None = None,
    force: bool | None = None,
)

Apply entire topology declaration.

Parameters:

• topology (Topology) –

  Topology to apply.

• consume (bool | None, default: None ) –

  If True, start consuming according to the topology.

• restore (bool | None, default: None ) –

  If True, restore on reconnect.

Source code in rmqaio/rmqaio.py

async def apply_topology(
    self,
    topology: Topology,
    consume: bool | None = None,
    restore: bool | None = None,
    force: bool | None = None,
):
    """
    Apply entire topology declaration.

    Args:
        topology: Topology to apply.
        consume: If `True`, start consuming according to the topology.
        restore: If `True`, restore on reconnect.
    """
    logger.info("apply topology[restore=%s] %s", topology, restore)

    for spec in topology.exchanges:
        await self.exchange_declare(spec, restore=restore, force=force)
    for spec in topology.queues:
        await self.queue_declare(spec, restore=restore, force=force)
    for spec in topology.bindings:
        await self.bind(spec, restore=restore)
    if consume:
        for spec in topology.consumers:
            if not next(filter(lambda consumer: consumer.spec == spec, self.consumers), None):
                await self.consume(spec, restore=restore)

check_exists async

check_exists(
    spec: BaseExchangeSpec | BaseQueueSpec,
    timeout: Number | None = None,
) -> bool

Check if subj exists.

Parameters:

• spec (BaseExchangeSpec | BaseQueueSpec) –

  Specification to check.

• timeout (Number | None, default: None ) –

  Operation timeout in seconds. If None, uses the default timeout.

Returns:

• bool –

  True if the subj exists, False otherwise.

Source code in rmqaio/rmqaio.py

async def check_exists(
    self,
    spec: BaseExchangeSpec | BaseQueueSpec,
    timeout: Number | None = None,
) -> bool:
    """
    Check if subj exists.

    Args:
        spec: Specification to check.
        timeout: Operation timeout in seconds. If `None`, uses the default timeout.

    Returns:
        True if the subj exists, False otherwise.
    """
    match spec:
        case BaseExchangeSpec():
            return await self.check_exchange_exists(spec.name, timeout=timeout)
        case BaseQueueSpec():
            return await self.check_queue_exists(spec.name, timeout=timeout)
        case _:
            raise ValueError("invalid spec type")

declare async

declare(
    spec: BaseExchangeSpec | BaseQueueSpec,
    timeout: Number | None = None,
    restore: bool | None = None,
    force: bool | None = None,
)

Declare subj.

Parameters:

• restore (bool | None, default: None ) –

  If True, automatically redeclare subj after reconnection.

• force (bool | None, default: None ) –

  If True, delete and redeclare subj if declaration fails due to parameter mismatch.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def declare(
    self,
    spec: BaseExchangeSpec | BaseQueueSpec,
    timeout: Number | None = None,
    restore: bool | None = None,
    force: bool | None = None,
):
    """
    Declare subj.

    Args:
        restore: If `True`, automatically redeclare subj after reconnection.
        force: If `True`, delete and redeclare subj if declaration fails
            due to parameter mismatch.
        timeout: Operation timeout. If `None`, uses the default timeout.
    """
    match spec:
        case BaseExchangeSpec():
            await self.exchange_declare(spec, timeout=timeout, restore=restore, force=force)
        case BaseQueueSpec():
            await self.queue_declare(spec, timeout=timeout, restore=restore, force=force)
        case _:
            raise ValueError("invalid spec type")

delete async

delete(spec: Spec, timeout: Number | None = None)

Delete subj.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Raises:

• Exception –

  If deleting fails.

Source code in rmqaio/rmqaio.py

async def delete(self, spec: Spec, timeout: Number | None = None):
    """
    Delete subj.

    Args:
        timeout: Operation timeout. If `None`, uses the default timeout.

    Raises:
        Exception: If deleting fails.
    """
    match spec:
        case BaseExchangeSpec():
            if spec.kind == "read-only":
                raise OperationError("can not delete read-only exchange")
            await self.exchange_delete(spec.name, timeout=timeout)
        case BaseQueueSpec():
            if spec.kind == "read-only":
                raise OperationError("can not delete read-only queue")
            await self.queue_delete(spec.name, timeout=timeout)
        case _:
            raise ValueError("invalid spec type")

check_exchange_exists async

check_exchange_exists(
    name: str, timeout: Number | None = None
) -> bool

Check if exchange exists.

Parameters:

• name (str) –

  Exchange name.

• timeout (Number | None, default: None ) –

  Operation timeout in seconds. If None, uses the default timeout.

Returns:

• bool –

  True if the exchange exists, False otherwise.

Source code in rmqaio/rmqaio.py

async def check_exchange_exists(self, name: str, timeout: Number | None = None) -> bool:
    """
    Check if exchange exists.

    Args:
        name: Exchange name.
        timeout: Operation timeout in seconds. If `None`, uses the default timeout.

    Returns:
        True if the exchange exists, False otherwise.
    """
    timeout_ = timeout if timeout is not None else self._timeout

    channel = await self._conn.channel(timeout=timeout_)

    try:
        await channel.exchange_declare(name, passive=True, timeout=timeout_)
        return True
    except aiormq.ChannelNotFoundEntity:
        return False

exchange_declare async

exchange_declare(
    spec: BaseExchangeSpec,
    timeout: Number | None = None,
    restore: bool | None = None,
    force: bool | None = None,
)

Declare the exchange.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds. If None, uses the default timeout.

• restore (bool | None, default: None ) –

  If True, automatically redeclare the exchange after reconnection.

• force (bool | None, default: None ) –

  If True, delete and redeclare the exchange if declaration fails due to parameter mismatch.

Raises:

• Exception –

  If declaring fails.

Source code in rmqaio/rmqaio.py

async def exchange_declare(
    self,
    spec: BaseExchangeSpec,
    timeout: Number | None = None,
    restore: bool | None = None,
    force: bool | None = None,
):
    """
    Declare the exchange.

    Args:
        timeout: Operation timeout in seconds. If `None`, uses the default timeout.
        restore: If `True`, automatically redeclare the exchange after reconnection.
        force: If `True`, delete and redeclare the exchange if declaration fails
            due to parameter mismatch.

    Raises:
        Exception: If declaring fails.
    """
    if spec.kind == "read-only":
        raise OperationError("can not declare read-only exchange")

    timeout_ = timeout if timeout is not None else self._timeout

    async def op():
        logger.info(_("declare[restore=%s, force=%s] %s"), restore, force, spec)

        channel = await self._conn.channel(timeout=timeout_)
        await channel.exchange_declare(
            spec.name,
            exchange_type=spec.type,
            durable=spec.durable,
            auto_delete=spec.auto_delete,
            arguments=spec.arguments.to_dict(),
            timeout=timeout_,
        )

    if force:

        async def on_error(e):
            channel = await self._conn.channel(timeout=timeout_)
            logger.info(_("delete[on_error] %s"), spec)
            await channel.exchange_delete(spec.name, timeout=timeout_)

        await retry(
            RetryPolicy(delays=[0], exc_filter=(aiormq.ChannelPreconditionFailed,)),
            on_error=on_error,
        )(op)()

    else:
        await op()

    if restore:
        self._topology.exchanges.append(spec)

exchange_delete async

exchange_delete(name: str, timeout: Number | None = None)

Delete exchange.

Parameters:

• name (str) –

  Exchange name.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def exchange_delete(self, name: str, timeout: Number | None = None):
    """
    Delete exchange.

    Args:
        name: Exchange name.
        timeout: Operation timeout. If `None`, uses the default timeout.
    """
    logger.info(_("delete exchange '%s'"), name)

    timeout_ = timeout if timeout is not None else self._timeout

    channel = await self._conn.channel(timeout=timeout_)

    await channel.exchange_delete(name, timeout=timeout_)

    spec = next(filter(lambda spec: spec.name == name, self._topology.exchanges), None)
    if spec:
        self._topology.exchanges.remove(spec)

check_queue_exists async

check_queue_exists(
    name: str, timeout: Number | None = None
) -> bool

Check if queue exists.

Parameters:

• name (str) –

  Queue name.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Returns:

• bool –

  True if the queue exists, False otherwise.

Source code in rmqaio/rmqaio.py

async def check_queue_exists(self, name: str, timeout: Number | None = None) -> bool:
    """
    Check if queue exists.

    Args:
        name: Queue name.
        timeout: Operation timeout. If `None`, uses the default timeout.

    Returns:
        True if the queue exists, False otherwise.
    """
    timeout_ = timeout if timeout is not None else self._timeout

    channel = await self._conn.channel(timeout=timeout_)

    try:
        await channel.queue_declare(name, passive=True, timeout=timeout_)
        return True
    except aiormq.ChannelNotFoundEntity:
        return False

queue_declare async

queue_declare(
    spec: BaseQueueSpec,
    timeout: Number | None = None,
    restore: bool | None = None,
    force: bool | None = None,
)

Declare queue.

Parameters:

• spec (BaseQueueSpec) –

  Queue specification.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

• restore (bool | None, default: None ) –

  Restore this binding on connection issue.

• force (bool | None, default: None ) –

  Force redeclare queue if it has already been declared with different parameters.

Raises:

• Exception –

  If declaring fails.

Source code in rmqaio/rmqaio.py

async def queue_declare(
    self,
    spec: BaseQueueSpec,
    timeout: Number | None = None,
    restore: bool | None = None,
    force: bool | None = None,
):
    """
    Declare queue.

    Args:
        spec: Queue specification.
        timeout: Operation timeout. If `None`, uses the default timeout.
        restore: Restore this binding on connection issue.
        force: Force redeclare queue if it has already been declared with different parameters.

    Raises:
        Exception: If declaring fails.
    """
    if spec.kind == "read-only":
        raise OperationError("can not declare read-only queue")

    timeout_ = timeout if timeout is not None else self._timeout

    async def op():
        logger.info(_("declare[restore=%s, force=%s] %s"), restore, force, spec)

        channel = await self._conn.channel(timeout=timeout_)
        arguments = spec.arguments.to_dict()
        await channel.queue_declare(
            spec.name,
            durable=spec.durable,
            exclusive=spec.exclusive,
            auto_delete=spec.auto_delete,
            arguments=arguments,
            timeout=timeout_,
        )

    if force:

        async def on_error(e):
            channel = await self._conn.channel(timeout=timeout_)
            logger.info(_("delete[on_error] %s"), spec)
            await channel.queue_delete(spec.name, timeout=timeout_)

        await retry(
            RetryPolicy(delays=[0], exc_filter=(aiormq.ChannelPreconditionFailed,)),
            on_error=on_error,
        )(op)()

    else:
        await op()

    if restore:
        self._topology.queues.append(spec)

queue_delete async

queue_delete(name: str, timeout: Number | None = None)

Delete queue.

Parameters:

• name (str) –

  Queue name.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def queue_delete(self, name: str, timeout: Number | None = None):
    """
    Delete queue.

    Args:
        name: Queue name.
        timeout: Operation timeout. If `None`, uses the default timeout.
    """
    logger.info(_("delete queue '%s'"), name)

    timeout_ = timeout if timeout is not None else self._timeout

    channel = await self._conn.channel(timeout=timeout_)

    await channel.queue_delete(name, timeout=timeout_)

    spec = next(filter(lambda spec: spec.name == name, self._topology.queues), None)
    if spec:
        self._topology.queues.remove(spec)

bind async

bind(
    spec: BindSpec,
    timeout: Number | None = None,
    restore: bool | None = None,
)

Bind item to exchange.

Parameters:

• spec (BindSpec) –

  Bind specification.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

• restore (bool | None, default: None ) –

  Restore this binding on connection issue.

Source code in rmqaio/rmqaio.py

async def bind(
    self,
    spec: BindSpec,
    timeout: Number | None = None,
    restore: bool | None = None,
):
    """
    Bind item to exchange.

    Args:
        spec: Bind specification.
        timeout: Operation timeout. If `None`, uses the default timeout.
        restore: Restore this binding on connection issue.
    """
    logger.info(
        _("bind %s '%s' to exchange '%s' with routing_key '%s'"),
        spec.kind,
        spec.dst,
        spec.src,
        spec.routing_key,
    )

    timeout_ = timeout if timeout is not None else self._timeout
    channel = await self._conn.channel(timeout=timeout_)

    match spec.kind:
        case "exchange":
            await channel.exchange_bind(
                spec.dst,
                spec.src,
                routing_key=spec.routing_key,
                timeout=timeout_,
            )
        case "queue":
            await channel.queue_bind(
                spec.dst,
                spec.src,
                routing_key=spec.routing_key,
                timeout=timeout_,
            )
        case _:
            raise ValueError("invalid spec type")

    if restore:
        self._topology.bindings.append(spec)

unbind async

unbind(spec: BindSpec, timeout: Number | None = None)

Unbind item from exchange.

Parameters:

• spec (BindSpec) –

  Bind specification.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def unbind(
    self,
    spec: BindSpec,
    timeout: Number | None = None,
):
    """
    Unbind item from exchange.

    Args:
        spec: Bind specification.
        timeout: Operation timeout. If `None`, uses the default timeout.
    """
    logger.info(
        _("unbind %s '%s' from exchange '%s' for routing_key '%s'"),
        spec.kind,
        spec.dst,
        spec.src,
        spec.routing_key,
    )

    timeout_ = timeout if timeout is not None else self._timeout
    channel = await self._conn.channel(timeout=timeout_)

    match spec.kind:
        case "exchange":
            await channel.exchange_unbind(
                spec.dst,
                spec.src,
                routing_key=spec.routing_key,
                timeout=timeout_,
            )
        case "queue":
            await channel.queue_unbind(
                spec.dst,
                spec.src,
                routing_key=spec.routing_key,
                timeout=timeout_,
            )
        case _:
            raise ValueError("invalid spec type")

    if spec in self._topology.bindings:
        self._topology.bindings.remove(spec)

publish async

publish(
    exchange: str,
    data: bytes,
    routing_key: str,
    properties: dict | None = None,
    mandatory: bool = False,
    timeout: Number | None = None,
)

Publish data to the exchange.

Parameters:

• exchange (str) –

  Exchange name.

• data (bytes) –

  Data to publish.

• routing_key (str) –

  Routing key for message delivery.

• properties (dict | None, default: None ) –

  Optional RabbitMQ message properties.

• mandatory (bool, default: False ) –

  If True, return unroutable message to publisher.

• timeout (Number | None, default: None ) –

  Operation timeout in seconds. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def publish(
    self,
    exchange: str,
    data: bytes,
    routing_key: str,
    properties: dict | None = None,
    mandatory: bool = False,
    timeout: Number | None = None,
):
    """
    Publish data to the exchange.

    Args:
        exchange: Exchange name.
        data: Data to publish.
        routing_key: Routing key for message delivery.
        properties: Optional RabbitMQ message properties.
        mandatory: If `True`, return unroutable message to publisher.
        timeout: Operation timeout in seconds. If `None`, uses the default timeout.
    """
    timeout_ = timeout if timeout is not None else self._timeout

    channel = await self._conn.channel(timeout=timeout_)

    if logger.isEnabledFor(logging.DEBUG):
        logger.debug(
            _("exchange[name='%s'] %s channel[%s] publish[routing_key='%s'] %s"),
            exchange,
            self._conn,
            channel,
            routing_key,
            (
                (
                    data[: config.log_data_truncate_size] + b"<truncated>"
                    if len(data) > config.log_data_truncate_size
                    else data
                )
                if not config.log_sanitize
                else "<hidden>"
            ),
        )

    await channel.basic_publish(
        data,
        exchange=exchange,
        routing_key=routing_key,
        properties=BasicProperties(**(properties or {})),
        mandatory=mandatory,
        timeout=timeout_,
    )

consume async

consume(
    spec: ConsumerSpec,
    timeout: Number | None = None,
    restore: bool | None = None,
) -> Consumer

Consume queue.

Parameters:

• spec (ConsumerSpec) –

  Spec.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

• restore (bool | None, default: None ) –

  Restore consuming on connection issue.

Returns:

• Consumer ( Consumer ) –

  The active consumer instance.

Source code in rmqaio/rmqaio.py

async def consume(
    self,
    spec: ConsumerSpec,
    timeout: Number | None = None,
    restore: bool | None = None,
) -> Consumer:
    """
    Consume queue.

    Args:
        spec: Spec.
        timeout: Operation timeout. If `None`, uses the default timeout.
        restore: Restore consuming on connection issue.

    Returns:
        Consumer: The active consumer instance.
    """
    timeout_ = timeout if timeout is not None else self._timeout

    channel = await self._conn.new_channel(timeout=timeout_)

    await channel.basic_qos(
        prefetch_count=spec.prefetch_count,
        prefetch_size=spec.prefetch_size,
        timeout=timeout_,
    )

    consumer_tag = cast(
        str,
        (
            await channel.basic_consume(
                spec.queue,
                partial(spec.callback, channel),
                no_ack=spec.auto_ack,
                exclusive=spec.exclusive,
                arguments=spec.arguments.to_dict(),
                consumer_tag=spec.consumer_tag,
                timeout=timeout_,
            )
        ).consumer_tag,
    )

    logger.info(_("consume %s"), spec)

    consumer = Consumer(spec, consumer_tag, channel)

    if restore:
        self._topology.consumers.append(spec)

    self._consumers[consumer_tag] = consumer

    return consumer

stop_consume async

stop_consume(
    consumer_tag: str | None = None,
    timeout: Number | None = None,
)

Stop consume.

Parameters:

• consumer_tag (str | None, default: None ) –

  Consumer tag. If None, stop all consumers.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def stop_consume(
    self,
    consumer_tag: str | None = None,
    timeout: Number | None = None,
):
    """
    Stop consume.

    Args:
        consumer_tag: Consumer tag. If `None`, stop all consumers.
        timeout: Operation timeout. If `None`, uses the default timeout.
    """
    if consumer_tag:
        tags = [consumer_tag]
    else:
        tags = list(self._consumers.keys())
    for tag in tags:
        if tag in self._consumers:
            consumer = self._consumers.pop(tag)
            if consumer.spec in self._topology.consumers:
                self._topology.consumers.remove(consumer.spec)
            if not consumer.channel.is_closed:
                logger.info(_("stop consume %s"), consumer.spec)
                timeout_ = timeout if timeout is not None else self._timeout
                await consumer.channel.basic_cancel(consumer.consumer_tag, timeout=timeout_)

DefaultExchange dataclass

Default exchange wrapper.

Attributes:

• spec (DefaultExchangeSpec) –

  Default exchange specification.

• ops (Ops) –

  Ops instance.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class DefaultExchange:
    """
    Default exchange wrapper.

    Attributes:
        spec: Default exchange specification.
        ops: Ops instance.
    """

    spec: DefaultExchangeSpec = field(init=False, default_factory=DefaultExchangeSpec)
    ops: Ops

    async def publish(
        self,
        data: bytes,
        routing_key: str,
        properties: dict | None = None,
        mandatory: bool = False,
        timeout: Number | None = None,
    ):
        """
        Publish data to the default exchange.

        Args:
            data: Data to publish.
            routing_key: Routing key for message delivery.
            properties: Optional RabbitMQ message properties.
            mandatory: If `True`, return unroutable message to publisher.
            timeout: Operation timeout in seconds. If `None`, uses the default timeout.
        """

        await self.ops.publish(
            self.spec.name,
            data,
            routing_key,
            properties=properties,
            mandatory=mandatory,
            timeout=timeout,
        )

publish async

publish(
    data: bytes,
    routing_key: str,
    properties: dict | None = None,
    mandatory: bool = False,
    timeout: Number | None = None,
)

Publish data to the default exchange.

Parameters:

• data (bytes) –

  Data to publish.

• routing_key (str) –

  Routing key for message delivery.

• properties (dict | None, default: None ) –

  Optional RabbitMQ message properties.

• mandatory (bool, default: False ) –

  If True, return unroutable message to publisher.

• timeout (Number | None, default: None ) –

  Operation timeout in seconds. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def publish(
    self,
    data: bytes,
    routing_key: str,
    properties: dict | None = None,
    mandatory: bool = False,
    timeout: Number | None = None,
):
    """
    Publish data to the default exchange.

    Args:
        data: Data to publish.
        routing_key: Routing key for message delivery.
        properties: Optional RabbitMQ message properties.
        mandatory: If `True`, return unroutable message to publisher.
        timeout: Operation timeout in seconds. If `None`, uses the default timeout.
    """

    await self.ops.publish(
        self.spec.name,
        data,
        routing_key,
        properties=properties,
        mandatory=mandatory,
        timeout=timeout,
    )

Exchange dataclass

Exchange wrapper.

Attributes:

• spec (BaseExchangeSpec) –

  Exchange specification.

• ops (Ops) –

  Ops instance.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class Exchange:
    """
    Exchange wrapper.

    Attributes:
        spec: Exchange specification.
        ops: Ops instance.
    """

    spec: BaseExchangeSpec
    ops: Ops

    async def check_exists(self, timeout: Number | None = None) -> bool:
        """
        Check if exchange exists.

        Args:
            timeout: Operation timeout in seconds. If `None`, uses the default timeout.

        Returns:
            True if the exchange exists, False otherwise.
        """
        return await self.ops.check_exchange_exists(self.spec.name, timeout=timeout)

    async def declare(
        self,
        timeout: Number | None = None,
        restore: bool | None = None,
        force: bool | None = None,
    ):
        """
        Declare exchange.

        Args:
            timeout: Operation timeout in seconds. If `None`, uses the default timeout.
            restore: Restore this exchange on connection issue.
            force: Force redeclare if already declared with different parameters.
        """
        await self.ops.exchange_declare(self.spec, timeout=timeout, restore=restore, force=force)

    async def delete(self, timeout: Number | None = None):
        """
        Delete exchange.

        Args:
            timeout: Operation timeout. If `None`, uses the default timeout.
        """
        await self.ops.delete(self.spec, timeout=timeout)

    async def bind(
        self,
        exchange: str,
        routing_key: str,
        timeout: Number | None = None,
        restore: bool | None = None,
    ):
        """
        Bind this exchange to another exchange.

        Args:
            exchange: Exchange name.
            routing_key: Routing key to bind.
            timeout: Operation timeout. If `None`, uses the default timeout.
            restore: Restore this binding on connection issue.
        """
        await self.ops.bind(
            BindSpec(src=exchange, dst=self.spec.name, routing_key=routing_key),
            timeout=timeout,
            restore=restore,
        )

    async def unbind(
        self,
        exchange: str,
        routing_key: str,
        timeout: Number | None = None,
    ):
        """
        Unbind this exchange from another exchange.

        Args:
            exchange: Exchange name.
            routing_key: Routing key to unbind.
            timeout: Operation timeout. If `None`, uses the default timeout.
        """
        await self.ops.unbind(
            BindSpec(src=exchange, dst=self.spec.name, routing_key=routing_key),
            timeout=timeout,
        )

    async def publish(
        self,
        data: bytes,
        routing_key: str,
        properties: dict | None = None,
        mandatory: bool = False,
        timeout: Number | None = None,
    ):
        """
        Publish data to the exchange.

        Args:
            data: Data to publish.
            routing_key: Routing key for message delivery.
            properties: Optional RabbitMQ message properties.
            mandatory: If `True`, return unroutable message to publisher.
            timeout: Operation timeout in seconds. If `None`, uses the default timeout.
        """

        await self.ops.publish(
            self.spec.name,
            data,
            routing_key,
            properties=properties,
            mandatory=mandatory,
            timeout=timeout,
        )

check_exists async

check_exists(timeout: Number | None = None) -> bool

Check if exchange exists.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds. If None, uses the default timeout.

Returns:

• bool –

  True if the exchange exists, False otherwise.

Source code in rmqaio/rmqaio.py

async def check_exists(self, timeout: Number | None = None) -> bool:
    """
    Check if exchange exists.

    Args:
        timeout: Operation timeout in seconds. If `None`, uses the default timeout.

    Returns:
        True if the exchange exists, False otherwise.
    """
    return await self.ops.check_exchange_exists(self.spec.name, timeout=timeout)

declare async

declare(
    timeout: Number | None = None,
    restore: bool | None = None,
    force: bool | None = None,
)

Declare exchange.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds. If None, uses the default timeout.

• restore (bool | None, default: None ) –

  Restore this exchange on connection issue.

• force (bool | None, default: None ) –

  Force redeclare if already declared with different parameters.

Source code in rmqaio/rmqaio.py

async def declare(
    self,
    timeout: Number | None = None,
    restore: bool | None = None,
    force: bool | None = None,
):
    """
    Declare exchange.

    Args:
        timeout: Operation timeout in seconds. If `None`, uses the default timeout.
        restore: Restore this exchange on connection issue.
        force: Force redeclare if already declared with different parameters.
    """
    await self.ops.exchange_declare(self.spec, timeout=timeout, restore=restore, force=force)

delete async

delete(timeout: Number | None = None)

Delete exchange.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def delete(self, timeout: Number | None = None):
    """
    Delete exchange.

    Args:
        timeout: Operation timeout. If `None`, uses the default timeout.
    """
    await self.ops.delete(self.spec, timeout=timeout)

bind async

bind(
    exchange: str,
    routing_key: str,
    timeout: Number | None = None,
    restore: bool | None = None,
)

Bind this exchange to another exchange.

Parameters:

• exchange (str) –

  Exchange name.

• routing_key (str) –

  Routing key to bind.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

• restore (bool | None, default: None ) –

  Restore this binding on connection issue.

Source code in rmqaio/rmqaio.py

async def bind(
    self,
    exchange: str,
    routing_key: str,
    timeout: Number | None = None,
    restore: bool | None = None,
):
    """
    Bind this exchange to another exchange.

    Args:
        exchange: Exchange name.
        routing_key: Routing key to bind.
        timeout: Operation timeout. If `None`, uses the default timeout.
        restore: Restore this binding on connection issue.
    """
    await self.ops.bind(
        BindSpec(src=exchange, dst=self.spec.name, routing_key=routing_key),
        timeout=timeout,
        restore=restore,
    )

unbind async

unbind(
    exchange: str,
    routing_key: str,
    timeout: Number | None = None,
)

Unbind this exchange from another exchange.

Parameters:

• exchange (str) –

  Exchange name.

• routing_key (str) –

  Routing key to unbind.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def unbind(
    self,
    exchange: str,
    routing_key: str,
    timeout: Number | None = None,
):
    """
    Unbind this exchange from another exchange.

    Args:
        exchange: Exchange name.
        routing_key: Routing key to unbind.
        timeout: Operation timeout. If `None`, uses the default timeout.
    """
    await self.ops.unbind(
        BindSpec(src=exchange, dst=self.spec.name, routing_key=routing_key),
        timeout=timeout,
    )

publish async

publish(
    data: bytes,
    routing_key: str,
    properties: dict | None = None,
    mandatory: bool = False,
    timeout: Number | None = None,
)

Publish data to the exchange.

Parameters:

• data (bytes) –

  Data to publish.

• routing_key (str) –

  Routing key for message delivery.

• properties (dict | None, default: None ) –

  Optional RabbitMQ message properties.

• mandatory (bool, default: False ) –

  If True, return unroutable message to publisher.

• timeout (Number | None, default: None ) –

  Operation timeout in seconds. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def publish(
    self,
    data: bytes,
    routing_key: str,
    properties: dict | None = None,
    mandatory: bool = False,
    timeout: Number | None = None,
):
    """
    Publish data to the exchange.

    Args:
        data: Data to publish.
        routing_key: Routing key for message delivery.
        properties: Optional RabbitMQ message properties.
        mandatory: If `True`, return unroutable message to publisher.
        timeout: Operation timeout in seconds. If `None`, uses the default timeout.
    """

    await self.ops.publish(
        self.spec.name,
        data,
        routing_key,
        properties=properties,
        mandatory=mandatory,
        timeout=timeout,
    )

Queue dataclass

Queue wrapper.

Attributes:

• spec (BaseQueueSpec) –

  Queue specification.

• ops (Ops) –

  Ops instance.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class Queue:
    """
    Queue wrapper.

    Attributes:
        spec: Queue specification.
        ops: Ops instance.
    """

    spec: BaseQueueSpec
    ops: Ops

    _consumers: dict[str, Consumer] = field(init=False, default_factory=dict)

    @property
    def consumers(self) -> list[Consumer]:
        return list(self._consumers.values())

    async def check_exists(self, timeout: Number | None = None) -> bool:
        """
        Check if queue exists.

        Args:
            timeout: Operation timeout in seconds. If `None`, uses the default timeout.

        Returns:
            True if the queue exists, False otherwise.
        """
        return await self.ops.check_queue_exists(self.spec.name, timeout=timeout)

    async def declare(
        self,
        timeout: Number | None = None,
        restore: bool | None = None,
        force: bool | None = None,
    ):
        """
        Declare queue.

        Args:
            timeout: Operation timeout in seconds. If `None`, uses the default timeout.
            restore: Restore this queue on connection issue.
            force: Force redeclare if already declared with different parameters.
        """
        await self.ops.queue_declare(self.spec, timeout=timeout, restore=restore, force=force)

    async def delete(self, timeout: Number | None = None):
        """
        Delete queue.

        Args:
            timeout: Operation timeout. If `None`, uses the default timeout.
        """
        await self.ops.delete(self.spec, timeout=timeout)

    async def bind(
        self,
        exchange: str,
        routing_key: str,
        timeout: Number | None = None,
        restore: bool | None = None,
    ):
        """
        Bind queue to exchange.

        Args:
            exchange: Exchange name.
            routing_key: Routing key to bind.
            timeout: Operation timeout. If `None`, uses the default timeout.
            restore: Restore this binding on connection issue.
        """
        await self.ops.bind(
            BindSpec(src=exchange, dst=self.spec.name, routing_key=routing_key),
            timeout=timeout,
            restore=restore,
        )

    async def unbind(
        self,
        exchange: str,
        routing_key: str,
        timeout: Number | None = None,
    ):
        """
        Unbind queue from exchange.

        Args:
            exchange: Exchange name.
            routing_key: Routing key to unbind.
            timeout: Operation timeout. If `None`, uses the default timeout.
        """
        await self.ops.unbind(
            BindSpec(src=exchange, dst=self.spec.name, routing_key=routing_key),
            timeout=timeout,
        )

    async def consume(
        self,
        callback: Callable[[aiormq.abc.AbstractChannel, aiormq.abc.DeliveredMessage], Awaitable],
        prefetch_count: int | None = None,
        prefetch_size: int | None = None,
        auto_ack: bool = True,
        exclusive: bool = False,
        consumer_tag: str | None = None,
        arguments: ConsumerArgs | None = None,
        timeout: Number | None = None,
        restore: bool | None = None,
    ) -> Consumer:
        """
        Start consuming messages from queue.

        Args:
            callback: Async callback function to handle messages.
            prefetch_count: Maximum number of unacknowledged messages.
            prefetch_size: Maximum number of unacknowledged bytes.
            auto_ack: If True, automatically acknowledge messages.
            exclusive: If True, create exclusive consumer.
            consumer_tag: Custom consumer tag.
            arguments: Consumer arguments.
            timeout: Operation timeout in seconds.
            restore: If True, restore consumer on reconnect.

        Returns:
            Consumer: Active consumer instance.
        """
        spec = ConsumerSpec(
            queue=self.spec.name,
            callback=callback,
            prefetch_count=prefetch_count,
            prefetch_size=prefetch_size,
            auto_ack=auto_ack,
            exclusive=exclusive,
            consumer_tag=consumer_tag,
            arguments=arguments or ConsumerArgs(),
        )
        consumer = await self.ops.consume(spec, timeout=timeout, restore=restore)
        self._consumers[consumer.consumer_tag] = consumer
        return consumer

    async def stop_consume(
        self,
        consumer_tag: str | None = None,
        timeout: Number | None = None,
    ):
        """
        Stop consume.

        Args:
            consumer_tag: Consumer tag. If `None`, stop all consumers.
            timeout: Operation timeout. If `None`, uses the default timeout.
        """
        if consumer_tag in self._consumers:
            consumer = self._consumers[consumer_tag]
            await self.ops.stop_consume(consumer_tag, timeout=timeout)
            self._consumers.pop(consumer.consumer_tag, None)
        else:
            for consumer in self.consumers:
                await self.ops.stop_consume(consumer_tag, timeout=timeout)
                self._consumers.pop(consumer.consumer_tag, None)

check_exists async

check_exists(timeout: Number | None = None) -> bool

Check if queue exists.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds. If None, uses the default timeout.

Returns:

• bool –

  True if the queue exists, False otherwise.

Source code in rmqaio/rmqaio.py

async def check_exists(self, timeout: Number | None = None) -> bool:
    """
    Check if queue exists.

    Args:
        timeout: Operation timeout in seconds. If `None`, uses the default timeout.

    Returns:
        True if the queue exists, False otherwise.
    """
    return await self.ops.check_queue_exists(self.spec.name, timeout=timeout)

declare async

declare(
    timeout: Number | None = None,
    restore: bool | None = None,
    force: bool | None = None,
)

Declare queue.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout in seconds. If None, uses the default timeout.

• restore (bool | None, default: None ) –

  Restore this queue on connection issue.

• force (bool | None, default: None ) –

  Force redeclare if already declared with different parameters.

Source code in rmqaio/rmqaio.py

async def declare(
    self,
    timeout: Number | None = None,
    restore: bool | None = None,
    force: bool | None = None,
):
    """
    Declare queue.

    Args:
        timeout: Operation timeout in seconds. If `None`, uses the default timeout.
        restore: Restore this queue on connection issue.
        force: Force redeclare if already declared with different parameters.
    """
    await self.ops.queue_declare(self.spec, timeout=timeout, restore=restore, force=force)

delete async

delete(timeout: Number | None = None)

Delete queue.

Parameters:

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def delete(self, timeout: Number | None = None):
    """
    Delete queue.

    Args:
        timeout: Operation timeout. If `None`, uses the default timeout.
    """
    await self.ops.delete(self.spec, timeout=timeout)

bind async

bind(
    exchange: str,
    routing_key: str,
    timeout: Number | None = None,
    restore: bool | None = None,
)

Bind queue to exchange.

Parameters:

• exchange (str) –

  Exchange name.

• routing_key (str) –

  Routing key to bind.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

• restore (bool | None, default: None ) –

  Restore this binding on connection issue.

Source code in rmqaio/rmqaio.py

async def bind(
    self,
    exchange: str,
    routing_key: str,
    timeout: Number | None = None,
    restore: bool | None = None,
):
    """
    Bind queue to exchange.

    Args:
        exchange: Exchange name.
        routing_key: Routing key to bind.
        timeout: Operation timeout. If `None`, uses the default timeout.
        restore: Restore this binding on connection issue.
    """
    await self.ops.bind(
        BindSpec(src=exchange, dst=self.spec.name, routing_key=routing_key),
        timeout=timeout,
        restore=restore,
    )

unbind async

unbind(
    exchange: str,
    routing_key: str,
    timeout: Number | None = None,
)

Unbind queue from exchange.

Parameters:

• exchange (str) –

  Exchange name.

• routing_key (str) –

  Routing key to unbind.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def unbind(
    self,
    exchange: str,
    routing_key: str,
    timeout: Number | None = None,
):
    """
    Unbind queue from exchange.

    Args:
        exchange: Exchange name.
        routing_key: Routing key to unbind.
        timeout: Operation timeout. If `None`, uses the default timeout.
    """
    await self.ops.unbind(
        BindSpec(src=exchange, dst=self.spec.name, routing_key=routing_key),
        timeout=timeout,
    )

consume async

consume(
    callback: Callable[
        [AbstractChannel, DeliveredMessage], Awaitable
    ],
    prefetch_count: int | None = None,
    prefetch_size: int | None = None,
    auto_ack: bool = True,
    exclusive: bool = False,
    consumer_tag: str | None = None,
    arguments: ConsumerArgs | None = None,
    timeout: Number | None = None,
    restore: bool | None = None,
) -> Consumer

Start consuming messages from queue.

Parameters:

• callback (Callable[[AbstractChannel, DeliveredMessage], Awaitable]) –

  Async callback function to handle messages.

• prefetch_count (int | None, default: None ) –

  Maximum number of unacknowledged messages.

• prefetch_size (int | None, default: None ) –

  Maximum number of unacknowledged bytes.

• auto_ack (bool, default: True ) –

  If True, automatically acknowledge messages.

• exclusive (bool, default: False ) –

  If True, create exclusive consumer.

• consumer_tag (str | None, default: None ) –

  Custom consumer tag.

• arguments (ConsumerArgs | None, default: None ) –

  Consumer arguments.

• timeout (Number | None, default: None ) –

  Operation timeout in seconds.

• restore (bool | None, default: None ) –

  If True, restore consumer on reconnect.

Returns:

• Consumer ( Consumer ) –

  Active consumer instance.

Source code in rmqaio/rmqaio.py

async def consume(
    self,
    callback: Callable[[aiormq.abc.AbstractChannel, aiormq.abc.DeliveredMessage], Awaitable],
    prefetch_count: int | None = None,
    prefetch_size: int | None = None,
    auto_ack: bool = True,
    exclusive: bool = False,
    consumer_tag: str | None = None,
    arguments: ConsumerArgs | None = None,
    timeout: Number | None = None,
    restore: bool | None = None,
) -> Consumer:
    """
    Start consuming messages from queue.

    Args:
        callback: Async callback function to handle messages.
        prefetch_count: Maximum number of unacknowledged messages.
        prefetch_size: Maximum number of unacknowledged bytes.
        auto_ack: If True, automatically acknowledge messages.
        exclusive: If True, create exclusive consumer.
        consumer_tag: Custom consumer tag.
        arguments: Consumer arguments.
        timeout: Operation timeout in seconds.
        restore: If True, restore consumer on reconnect.

    Returns:
        Consumer: Active consumer instance.
    """
    spec = ConsumerSpec(
        queue=self.spec.name,
        callback=callback,
        prefetch_count=prefetch_count,
        prefetch_size=prefetch_size,
        auto_ack=auto_ack,
        exclusive=exclusive,
        consumer_tag=consumer_tag,
        arguments=arguments or ConsumerArgs(),
    )
    consumer = await self.ops.consume(spec, timeout=timeout, restore=restore)
    self._consumers[consumer.consumer_tag] = consumer
    return consumer

stop_consume async

stop_consume(
    consumer_tag: str | None = None,
    timeout: Number | None = None,
)

Stop consume.

Parameters:

• consumer_tag (str | None, default: None ) –

  Consumer tag. If None, stop all consumers.

• timeout (Number | None, default: None ) –

  Operation timeout. If None, uses the default timeout.

Source code in rmqaio/rmqaio.py

async def stop_consume(
    self,
    consumer_tag: str | None = None,
    timeout: Number | None = None,
):
    """
    Stop consume.

    Args:
        consumer_tag: Consumer tag. If `None`, stop all consumers.
        timeout: Operation timeout. If `None`, uses the default timeout.
    """
    if consumer_tag in self._consumers:
        consumer = self._consumers[consumer_tag]
        await self.ops.stop_consume(consumer_tag, timeout=timeout)
        self._consumers.pop(consumer.consumer_tag, None)
    else:
        for consumer in self.consumers:
            await self.ops.stop_consume(consumer_tag, timeout=timeout)
            self._consumers.pop(consumer.consumer_tag, None)

retry

retry(
    retry_policy: RetryPolicy,
    *,
    msg: str | None = None,
    on_error: Callable[[Exception], Awaitable] | None = None
)

Create a retry decorator for async functions.

The decorated coroutine will be retried using the provided delay sequence if the raised exception matches exc_filter.

Parameters:

• retry_policy (RetryPolicy) –

  Retry policy.

• msg (str | None, default: None ) –

  Optional message to log before retrying. If not provided, the function object is logged.

• on_error (Callable[[Exception], Awaitable] | None, default: None ) –

  Optional async callback executed after a retryable exception is caught and before sleeping.

Returns:

• –

  Decorator for async functions.

Source code in rmqaio/rmqaio.py

def retry(
    retry_policy: RetryPolicy,
    *,
    msg: str | None = None,
    on_error: Callable[[Exception], Awaitable] | None = None,
):
    """
    Create a retry decorator for async functions.

    The decorated coroutine will be retried using the provided delay
    sequence if the raised exception matches `exc_filter`.

    Args:
        retry_policy: Retry policy.
        msg: Optional message to log before retrying. If not provided,
            the function object is logged.
        on_error: Optional async callback executed after a retryable
            exception is caught and before sleeping.

    Returns:
        Decorator for async functions.
    """

    def decorator(fn):
        @wraps(fn)
        async def wrapper(*args, **kwds):
            timeouts = iter(retry_policy.delays)
            attempt = 0
            while True:
                try:
                    return await fn(*args, **kwds)
                except Exception as e:
                    if not retry_policy.is_retryable(e):
                        raise
                    try:
                        t = next(timeouts)
                        attempt += 1
                        logger.warning(
                            _("%s (%s %s) retry(%s) in %s second(s)"),
                            msg or fn,
                            e.__class__,
                            e,
                            attempt,
                            t,
                        )
                        if on_error:
                            await on_error(e)
                        await sleep(t)
                    except StopIteration:
                        raise e  # raise original exception instead of StopIteration

        return wrapper

    return decorator