API reference

rmqaio

Number module-attribute

Number = int | float

Numeric type alias for integers or floats.

RmqAioError

Bases: Exception

Base exception for all rmqaio errors.

Source code in rmqaio/rmqaio.py

class RmqAioError(Exception):
    """Base exception for all rmqaio errors."""

ConnectionInvalidStateError

Bases: RmqAioError

Raised when an operation is attempted in an invalid connection state.

Source code in rmqaio/rmqaio.py

class ConnectionInvalidStateError(RmqAioError):
    """Raised when an operation is attempted in an invalid connection state."""

OperationError

Bases: RmqAioError

Raised when an operation is not allowed on a read-only entity.

Source code in rmqaio/rmqaio.py

class OperationError(RmqAioError):
    """Raised when an operation is not allowed on a read-only entity."""

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: object) -> bool:
        if not isinstance(other, Repeat):
            return NotImplemented
        return self.value == other.value

    def __ne__(self, other: object) -> bool:
        if not isinstance(other, Repeat):
            return NotImplemented
        return 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: str, Enum

Enum for representing the state of a connection.

Source code in rmqaio/rmqaio.py

class ConnectionState(str, Enum):
    """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."""

    RECONNECTING = "reconnecting"
    """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.

RECONNECTING class-attribute instance-attribute

RECONNECTING = 'reconnecting'

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._state_lock = _ReentrantLock()
        self._channel_lock = Lock()
        self._conn: aiormq.Connection | None = None
        self._channel: aiormq.abc.AbstractChannel | None = None
        self._open_futures: set[asyncio.Future] = set()
        self._refresh_futures: set[asyncio.Future] = set()
        self._closed_event = asyncio.Event()
        self._loop_task: asyncio.Task | None = None
        self._exc: BaseException | 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 is not None:
            return value // 1000  # milliseconds to seconds
        return None

    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: type[BaseException] | None, exc: BaseException | None, tb: object):
        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:
        """Policy for first connection attempts. `None` means no retries."""
        return self._open_retry_policy

    @property
    def reopen_retry_policy(self) -> RetryPolicy | None:
        """Policy for reconnection attempts. `None` means no retries."""
        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 or in the process of closing."""
        return self._state in [ConnectionState.CLOSED, ConnectionState.CLOSING]

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

    async def _open(self, timeout: Number | None = None):
        loop = get_running_loop()
        future = loop.create_future()
        self._open_futures.add(future)
        try:
            if self._state == ConnectionState.INITIAL:
                if not self._loop_task:
                    self._loop_task = asyncio.create_task(self._loop())
            await asyncio.wait_for(future, timeout=timeout)
        finally:
            self._open_futures.discard(future)

    async def _refresh(self, timeout: Number | None = None):
        loop = get_running_loop()
        future = loop.create_future()
        self._refresh_futures.add(future)
        try:
            await self._set_state(ConnectionState.REFRESHING)
            await cast(aiormq.Connection, self._conn).close()
            await asyncio.wait_for(future, timeout=timeout)
        finally:
            self._refresh_futures.discard(future)

    async def _close(self, timeout: Number | None = None):
        await self._set_state(ConnectionState.CLOSING)
        if self._loop_task:
            if asyncio.current_task() is self._loop_task:
                if self._conn and not self._conn.is_closed:
                    await self._conn.close()
            else:
                self._loop_task.cancel()
        if self._loop_task and asyncio.current_task() is not 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. Reconnection on connection
        loss is handled automatically by the internal loop.

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

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

        await self._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._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._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.
        """
        async with self._channel_lock:
            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.

        If the callback is a bound method, it is stored via `weakref.WeakMethod` so that
        the referenced object can be garbage-collected. When the object is collected, the
        wrapper automatically removes itself on the next invocation.
        """
        if hasattr(callback, "__self__"):
            ref = weakref.WeakMethod(callback)

            async def weak_wrapper(state_from, state_to):
                cb = ref()
                if cb is None:
                    self._callbacks.pop(name, None)
                    return
                await cb(state_from, state_to)

            self._callbacks[name] = weak_wrapper
        else:
            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 executing 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):
        async with self._state_lock:
            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 _connect(self):
        retry_policy = self._open_retry_policy if self._state == ConnectionState.INITIAL else self._reopen_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)

                if self._state == ConnectionState.INITIAL:
                    await self._set_state(ConnectionState.CONNECTING)
                else:
                    await self._set_state(ConnectionState.RECONNECTING)

                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

                for f in list(self._open_futures):
                    if not f.done():
                        f.set_result(None)
                for f in list(self._refresh_futures):
                    if not f.done():
                        f.set_result(None)

                await self._set_state(ConnectionState.CONNECTED)

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

            except asyncio.CancelledError:
                raise

            except Exception as e:
                self._exc = e

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

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

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

                await sleep(delay)

    async def _monitor(self):
        try:
            while True:
                done = (await wait([self._conn.closing], timeout=5, return_when=FIRST_COMPLETED))[0]
                if done or (self._conn and self._conn.is_connection_was_stuck):
                    break

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

        except asyncio.CancelledError:
            raise

    async def _loop(self):
        try:
            while self._state not in [ConnectionState.CLOSING, ConnectionState.CLOSED]:
                await self._connect()

                if self._state is not ConnectionState.CONNECTED:
                    break

                await self._monitor()

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

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

        except Exception as e:
            self._exc = e
            logger.exception(e)

        finally:
            exc = self._exc
            if isinstance(exc, CancelledError):
                exc = None
            exc = exc or ConnectionInvalidStateError("connection closed")

            for f in list(self._open_futures):
                if not f.done():
                    f.set_exception(exc)
            self._open_futures.clear()
            for f in list(self._refresh_futures):
                if not f.done():
                    f.set_exception(exc)
            self._refresh_futures.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

Policy for first connection attempts. None means no retries.

reopen_retry_policy property

reopen_retry_policy: RetryPolicy | None

Policy for reconnection attempts. None means no retries.

is_open property

is_open: bool

Whether connection is open and operational.

is_closed property

is_closed: bool

Whether connection is closed or in the process of closing.

__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._state_lock = _ReentrantLock()
    self._channel_lock = Lock()
    self._conn: aiormq.Connection | None = None
    self._channel: aiormq.abc.AbstractChannel | None = None
    self._open_futures: set[asyncio.Future] = set()
    self._refresh_futures: set[asyncio.Future] = set()
    self._closed_event = asyncio.Event()
    self._loop_task: asyncio.Task | None = None
    self._exc: BaseException | 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. Reconnection on connection loss is handled automatically by the internal loop.

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. Reconnection on connection
    loss is handled automatically by the internal loop.

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

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

    await self._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._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._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.
    """
    async with self._channel_lock:
        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.

If the callback is a bound method, it is stored via weakref.WeakMethod so that the referenced object can be garbage-collected. When the object is collected, the wrapper automatically removes itself on the next invocation.

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.

    If the callback is a bound method, it is stored via `weakref.WeakMethod` so that
    the referenced object can be garbage-collected. When the object is collected, the
    wrapper automatically removes itself on the next invocation.
    """
    if hasattr(callback, "__self__"):
        ref = weakref.WeakMethod(callback)

        async def weak_wrapper(state_from, state_to):
            cb = ref()
            if cb is None:
                self._callbacks.pop(name, None)
                return
            await cb(state_from, state_to)

        self._callbacks[name] = weak_wrapper
    else:
        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_lock = Lock()
        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: type[BaseException] | None, exc: BaseException | None, tb: object):
        await self.close()

    @property
    def connection(self) -> Connection:
        """Underlying Connection instance shared by all references."""
        return self._conn

    @classmethod
    def get_all_connections(cls) -> list[Connection]:
        """Return all unique underlying Connection instances."""
        return list(set(item.conn for item in cls._shared.values()))

    @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:
        """Policy for first connection attempts. `None` means no retries."""
        return self._conn.open_retry_policy

    @property
    def reopen_retry_policy(self) -> RetryPolicy | None:
        """Policy for reconnection attempts. `None` means no retries."""
        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. Reconnection on connection
        loss is handled automatically by the internal loop.

        Args:
            timeout: Operation timeout in seconds.

        Raises:
            ConnectionInvalidStateError: If the connection has already been closed.
        """
        async with self._lock:
            if self._is_closed.is_set():
                raise ConnectionInvalidStateError(_("can not reopen closed connection"))

            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 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.
        """
        async with self._channel_lock:
            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)

connection property

connection: Connection

Underlying Connection instance shared by all references.

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

Policy for first connection attempts. None means no retries.

reopen_retry_policy property

reopen_retry_policy: RetryPolicy | None

Policy for reconnection attempts. None means no retries.

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_lock = Lock()
    self._channel: aiormq.abc.AbstractChannel | None = None
    self._is_open = asyncio.Event()
    self._is_closed = asyncio.Event()

get_all_connections classmethod

get_all_connections() -> list[Connection]

Return all unique underlying Connection instances.

Source code in rmqaio/rmqaio.py

@classmethod
def get_all_connections(cls) -> list[Connection]:
    """Return all unique underlying Connection instances."""
    return list(set(item.conn for item in cls._shared.values()))

open async

open(timeout: Number | None = None)

Open connection to RabbitMQ.

Establishes connection to RabbitMQ broker. Reconnection on connection loss is handled automatically by the internal loop.

Parameters:

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

  Operation timeout in seconds.

Raises:

• ConnectionInvalidStateError –

  If the connection has already been closed.

Source code in rmqaio/rmqaio.py

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

    Establishes connection to RabbitMQ broker. Reconnection on connection
    loss is handled automatically by the internal loop.

    Args:
        timeout: Operation timeout in seconds.

    Raises:
        ConnectionInvalidStateError: If the connection has already been closed.
    """
    async with self._lock:
        if self._is_closed.is_set():
            raise ConnectionInvalidStateError(_("can not reopen closed connection"))

        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 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.
    """
    async with self._channel_lock:
        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)

Spec

Bases: Protocol

Protocol for entities that have a kind and a name.

Source code in rmqaio/rmqaio.py

class Spec(Protocol):
    """Protocol for entities that have a kind and a name."""

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

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

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=''"))

DelayedExchangeArgs dataclass

Bases: BaseExchangeArgs

Arguments for a delayed exchange.

Attributes:

• delayed_type (DelayedExchangeType) –

  Underlying exchange type for the delayed message plugin.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class DelayedExchangeArgs(BaseExchangeArgs):
    """
    Arguments for a delayed exchange.

    Attributes:
        delayed_type: Underlying exchange type for the delayed message plugin.
    """

    delayed_type: DelayedExchangeType = "direct"

    def to_dict(self) -> dict[str, Any]:
        """
        Convert arguments to a dictionary.

        Returns:
            Dictionary with x-delayed-type and inherited arguments.
        """
        args = BaseExchangeArgs.to_dict(self)
        args["x-delayed-type"] = self.delayed_type
        return args

to_dict

to_dict() -> dict[str, Any]

Convert arguments to a dictionary.

Returns:

• dict[str, Any] –

  Dictionary with x-delayed-type and inherited arguments.

Source code in rmqaio/rmqaio.py

def to_dict(self) -> dict[str, Any]:
    """
    Convert arguments to a dictionary.

    Returns:
        Dictionary with x-delayed-type and inherited arguments.
    """
    args = BaseExchangeArgs.to_dict(self)
    args["x-delayed-type"] = self.delayed_type
    return args

DelayedExchangeSpec dataclass

Bases: BaseExchangeSpec

Specification for a delayed exchange (x-delayed-message plugin).

Attributes:

• name (str) –

  Exchange name.

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

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

• type (Literal['x-delayed-message']) –

  Always "x-delayed-message".

• durable (bool) –

  Whether exchange is durable.

• auto_delete (bool) –

  Whether to delete when no longer used.

• internal (bool) –

  Whether exchange is internal.

• arguments (DelayedExchangeArgs) –

  Delayed exchange arguments.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class DelayedExchangeSpec(BaseExchangeSpec):
    """
    Specification for a delayed exchange (x-delayed-message plugin).

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

    name: str
    kind: Literal["normal", "read-only"] = field(default="normal")
    type: Literal["x-delayed-message"] = field(init=False, default="x-delayed-message")
    durable: bool = True
    auto_delete: bool = False
    internal: bool = False

    arguments: DelayedExchangeArgs = field(default_factory=DelayedExchangeArgs)

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

BaseQueueArgs dataclass

Base queue arguments.

Attributes:

• queue_type (QueueType | None) –

  Queue type (classic, quorum, or stream).

• message_ttl (int | None) –

  Time-to-live for messages in milliseconds.

• expires (int | None) –

  Queue expiry in milliseconds after last use.

• max_length (int | None) –

  Maximum number of messages in the queue.

• max_length_bytes (int | None) –

  Maximum queue size in bytes.

• overflow (OverflowPolicy | None) –

  Overflow behaviour when max length is reached.

• dead_letter_exchange (str | None) –

  Exchange to route dead letters to.

• dead_letter_routing_key (str | None) –

  Routing key for dead letters.

• max_priority (int | None) –

  Maximum priority for priority queues.

• single_active_consumer (bool | None) –

  Enables single active consumer.

• queue_mode (QueueMode | None) –

  Queue mode (default or lazy).

• delivery_limit (int | None) –

  Maximum number of delivery attempts.

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

  Custom x-arguments for the queue.

Source code in rmqaio/rmqaio.py

@dataclass(frozen=True, slots=True)
class BaseQueueArgs:
    """
    Base queue arguments.

    Attributes:
        queue_type: Queue type (classic, quorum, or stream).
        message_ttl: Time-to-live for messages in milliseconds.
        expires: Queue expiry in milliseconds after last use.
        max_length: Maximum number of messages in the queue.
        max_length_bytes: Maximum queue size in bytes.
        overflow: Overflow behaviour when max length is reached.
        dead_letter_exchange: Exchange to route dead letters to.
        dead_letter_routing_key: Routing key for dead letters.
        max_priority: Maximum priority for priority queues.
        single_active_consumer: Enables single active consumer.
        queue_mode: Queue mode (default or lazy).
        delivery_limit: Maximum number of delivery attempts.
        custom: Custom x-arguments for the queue.
    """

    queue_type: QueueType | None = None
    message_ttl: int | None = None
    expires: int | None = None
    max_length: int | None = None
    max_length_bytes: int | None = None
    overflow: OverflowPolicy | None = None
    dead_letter_exchange: str | None = None
    dead_letter_routing_key: str | None = None
    max_priority: int | None = None
    single_active_consumer: bool | None = None
    queue_mode: QueueMode | None = None
    delivery_limit: int | None = None
    custom: dict[str, Any] | None = None

    def to_dict(self) -> dict[str, Any]:
        """
        Convert queue arguments to a dictionary of AMQP x-arguments.

        Returns:
            Dictionary with x-* keys for RabbitMQ queue declaration.
        """
        args: dict[str, Any] = {}

        if self.queue_type is not None:
            args["x-queue-type"] = self.queue_type

        if self.message_ttl is not None:
            args["x-message-ttl"] = self.message_ttl

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

        if self.max_length is not None:
            args["x-max-length"] = self.max_length

        if self.max_length_bytes is not None:
            args["x-max-length-bytes"] = self.max_length_bytes

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

        if self.dead_letter_exchange is not None:
            args["x-dead-letter-exchange"] = self.dead_letter_exchange

        if self.dead_letter_routing_key is not None:
            args["x-dead-letter-routing-key"] = self.dead_letter_routing_key

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

        if self.single_active_consumer is not None:
            args["x-single-active-consumer"] = self.single_active_consumer

        if self.queue_mode is not None:
            args["x-queue-mode"] = self.queue_mode

        if self.delivery_limit is not None:
            args["x-delivery-limit"] = self.delivery_limit

        if self.custom:
            args.update(self.custom)

        return args

to_dict

to_dict() -> dict[str, Any]

Convert queue arguments to a dictionary of AMQP x-arguments.

Returns:

• dict[str, Any] –

  Dictionary with x-* keys for RabbitMQ queue declaration.

Source code in rmqaio/rmqaio.py

def to_dict(self) -> dict[str, Any]:
    """
    Convert queue arguments to a dictionary of AMQP x-arguments.

    Returns:
        Dictionary with x-* keys for RabbitMQ queue declaration.
    """
    args: dict[str, Any] = {}

    if self.queue_type is not None:
        args["x-queue-type"] = self.queue_type

    if self.message_ttl is not None:
        args["x-message-ttl"] = self.message_ttl

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

    if self.max_length is not None:
        args["x-max-length"] = self.max_length

    if self.max_length_bytes is not None:
        args["x-max-length-bytes"] = self.max_length_bytes

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

    if self.dead_letter_exchange is not None:
        args["x-dead-letter-exchange"] = self.dead_letter_exchange

    if self.dead_letter_routing_key is not None:
        args["x-dead-letter-routing-key"] = self.dead_letter_routing_key

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

    if self.single_active_consumer is not None:
        args["x-single-active-consumer"] = self.single_active_consumer

    if self.queue_mode is not None:
        args["x-queue-mode"] = self.queue_mode

    if self.delivery_limit is not None:
        args["x-delivery-limit"] = self.delivery_limit

    if self.custom:
        args.update(self.custom)

    return args

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"

to_dict

to_dict() -> dict[str, Any]

Convert queue arguments to a dictionary of AMQP x-arguments.

Returns:

• dict[str, Any] –

  Dictionary with x-* keys for RabbitMQ queue declaration.

Source code in rmqaio/rmqaio.py

def to_dict(self) -> dict[str, Any]:
    """
    Convert queue arguments to a dictionary of AMQP x-arguments.

    Returns:
        Dictionary with x-* keys for RabbitMQ queue declaration.
    """
    args: dict[str, Any] = {}

    if self.queue_type is not None:
        args["x-queue-type"] = self.queue_type

    if self.message_ttl is not None:
        args["x-message-ttl"] = self.message_ttl

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

    if self.max_length is not None:
        args["x-max-length"] = self.max_length

    if self.max_length_bytes is not None:
        args["x-max-length-bytes"] = self.max_length_bytes

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

    if self.dead_letter_exchange is not None:
        args["x-dead-letter-exchange"] = self.dead_letter_exchange

    if self.dead_letter_routing_key is not None:
        args["x-dead-letter-routing-key"] = self.dead_letter_routing_key

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

    if self.single_active_consumer is not None:
        args["x-single-active-consumer"] = self.single_active_consumer

    if self.queue_mode is not None:
        args["x-queue-mode"] = self.queue_mode

    if self.delivery_limit is not None:
        args["x-delivery-limit"] = self.delivery_limit

    if self.custom:
        args.update(self.custom)

    return args

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], Coroutine[Any, Any, Any]]) –

  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], Coroutine[Any, Any, Any]]
    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

UniqueList

Bases: MutableSequence[T], Generic[T]

A list that enforces uniqueness of its elements.

Maintains insertion order and uses a dict internally for O(1) membership tests. Only hashable items are supported.

Examples:

>>> ul = UniqueList([1, 2, 2, 3])
>>> list(ul)
[1, 2, 3]

Source code in rmqaio/rmqaio.py

class UniqueList(MutableSequence[T], Generic[T]):
    """
    A list that enforces uniqueness of its elements.

    Maintains insertion order and uses a dict internally for O(1) membership
    tests. Only hashable items are supported.

    Examples:
        >>> ul = UniqueList([1, 2, 2, 3])
        >>> list(ul)
        [1, 2, 3]
    """

    def __init__(self, iterable: Iterable[T] | None = None):
        self._data: dict[T, None] = {}
        if iterable is not None:
            for item in iterable:
                self.append(item)

    def __len__(self):
        return len(self._data)

    @overload
    def __getitem__(self, index: int) -> T: ...

    @overload
    def __getitem__(self, index: slice) -> list[T]: ...

    def __getitem__(self, index):
        keys = list(self._data)
        if isinstance(index, slice):
            return keys[index]
        return keys[index]

    @overload
    def __setitem__(self, index: int, value: T) -> None: ...

    @overload
    def __setitem__(self, index: slice, value: Iterable[T]) -> None: ...

    def __setitem__(self, index: int | slice, value: T | Iterable[T]) -> None:
        if isinstance(index, slice):
            raise TypeError("slice assignment is not supported")

        value = cast(T, value)
        keys = list(self._data)

        keys.pop(index)

        try:
            keys.remove(value)
        except ValueError:
            pass

        keys.insert(index, value)
        self._data = dict.fromkeys(keys)

    @overload
    def __delitem__(self, index: int) -> None: ...

    @overload
    def __delitem__(self, index: slice) -> None: ...

    def __delitem__(self, index: int | slice) -> None:
        keys = list(self._data)
        del keys[index]
        self._data = dict.fromkeys(keys)

    def insert(self, index: int, value: T):
        keys = list(self._data)

        try:
            keys.remove(value)
        except ValueError:
            pass

        keys.insert(index, value)
        self._data = dict.fromkeys(keys)

    def append(self, value: T):
        self._data[value] = None

    def remove(self, value: T):
        try:
            del self._data[value]
        except KeyError:
            raise ValueError(f"{value!r} not in list")

    def __contains__(self, value):
        return value in self._data

    def __iter__(self):
        return iter(self._data)

    def copy(self):
        return self.__class__(self)

    def __eq__(self, other):
        if isinstance(other, MutableSequence):
            return list(self) == list(other)
        return NotImplemented

    def __repr__(self):
        return f"{self.__class__.__name__}({list(self._data)})"

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=UniqueList[BaseExchangeSpec])
    queues: UniqueList[BaseQueueSpec] = field(default_factory=UniqueList[BaseQueueSpec])
    bindings: UniqueList[BindSpec] = field(default_factory=UniqueList[BindSpec])
    consumers: UniqueList[ConsumerSpec] = field(default_factory=UniqueList[ConsumerSpec])

Ops

RabbitMQ operations handler.

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

Attributes:

• conn (ConnectionProtocol) –

  Connection to RabbitMQ.

• timeout (ConnectionProtocol) –

  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._restore_task: asyncio.Task | None = None

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

    @property
    def conn(self) -> ConnectionProtocol:
        """Connection to RabbitMQ."""
        return self._conn

    async def _on_connection_state_changed(self, state_from: ConnectionState, state_to: ConnectionState):
        if state_to in (ConnectionState.CLOSING, ConnectionState.CLOSED):
            if self._restore_task and not self._restore_task.done():
                self._restore_task.cancel()
                self._restore_task = None
        elif state_to == ConnectionState.CONNECTED and state_from != ConnectionState.CONNECTING:
            if self._restore_task and not self._restore_task.done():
                self._restore_task.cancel()
            self._restore_task = asyncio.create_task(self._restore_topology())

    @retry(
        RetryPolicy(
            delays=Repeat(5),
            exc_filter=(
                asyncio.TimeoutError,
                ConnectionError,
                aiormq.exceptions.AMQPConnectionError,
                aiormq.exceptions.ChannelLockedResource,
            ),
        ),
        msg="restore topology",
    )
    async def _restore_topology(self):
        if not self._conn.is_open or self._conn.is_closed:
            raise ConnectionInvalidStateError(_("connection lost during restore"))

        for consumer in list(self._consumers.values()):
            if not consumer.channel.is_closed:
                try:
                    await consumer.channel.close()
                except Exception:
                    pass
        self._consumers.clear()

        for spec in self._topology.exchanges:
            await self.exchange_declare(spec, restore=True)

        for spec in self._topology.queues:
            await retry(
                RetryPolicy(
                    delays=[0.05, 0.05, 0.05],
                    exc_filter=(aiormq.exceptions.ChannelLockedResource,),
                )
            )(
                self.queue_declare
            )(spec, restore=True)

        for spec in self._topology.bindings:
            await self.bind(spec, restore=True)

        for spec in self._topology.consumers:
            await self.consume(spec, restore=True)

    @property
    def consumers(self) -> list[Consumer]:
        """Currently active consumers."""
        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.
            force: If `True`, delete and redeclare on parameter mismatch.
        """
        logger.info(_("applying topology[restore=%s] %s"), restore, topology)

        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((consumer for consumer in self.consumers if consumer.spec == spec), 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:
            spec: Specification of the exchange or queue to declare.
            timeout: Operation timeout. If `None`, uses the default timeout.
            restore: If `True`, automatically redeclare subj after reconnection.
            force: If `True`, delete and redeclare subj if declaration fails
                due to parameter mismatch.
        """
        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(_("declaring[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(_("deleting[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(_("deleting 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 get_queue(self, name: str, timeout: Number | None = None) -> QueueDeclareOk:
        """
        Get queue declare info.

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

        Returns:
            Queue declare OK result from the broker.
        """
        timeout_ = timeout if timeout is not None else self._timeout

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

        return await channel.queue_declare(name, passive=True, timeout=timeout_)

    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(_("declaring[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(_("deleting[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(_("deleting 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(
            _("binding[restore=%s] %s '%s' to exchange '%s' with routing_key '%s'"),
            restore,
            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(
            _("unbinding %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[str, Any] | 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] publishing[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,
                    lambda msg: spec.callback(channel, msg),
                    no_ack=spec.auto_ack,
                    exclusive=spec.exclusive,
                    arguments=spec.arguments.to_dict(),
                    consumer_tag=spec.consumer_tag,
                    timeout=timeout_,
                )
            ).consumer_tag,
        )

        logger.info(_("consuming[restore=%s] %s"), restore, 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 consuming %s"), consumer.spec)
                    timeout_ = timeout if timeout is not None else self._timeout
                    await consumer.channel.basic_cancel(consumer.consumer_tag, timeout=timeout_)

conn property

conn: ConnectionProtocol

Connection to RabbitMQ.

consumers property

consumers: list[Consumer]

Currently active consumers.

__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._restore_task: asyncio.Task | None = None

    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.

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

  If True, delete and redeclare on parameter mismatch.

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.
        force: If `True`, delete and redeclare on parameter mismatch.
    """
    logger.info(_("applying topology[restore=%s] %s"), restore, topology)

    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((consumer for consumer in self.consumers if consumer.spec == spec), 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:

• spec (BaseExchangeSpec | BaseQueueSpec) –

  Specification of the exchange or queue to declare.

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

  Operation timeout. If None, uses the default timeout.

• 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.

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:
        spec: Specification of the exchange or queue to declare.
        timeout: Operation timeout. If `None`, uses the default timeout.
        restore: If `True`, automatically redeclare subj after reconnection.
        force: If `True`, delete and redeclare subj if declaration fails
            due to parameter mismatch.
    """
    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(_("declaring[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(_("deleting[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(_("deleting 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

get_queue async

get_queue(
    name: str, timeout: Number | None = None
) -> QueueDeclareOk

Get queue declare info.

Parameters:

• name (str) –

  Queue name.

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

  Operation timeout. If None, uses the default timeout.

Returns:

• QueueDeclareOk –

  Queue declare OK result from the broker.

Source code in rmqaio/rmqaio.py

async def get_queue(self, name: str, timeout: Number | None = None) -> QueueDeclareOk:
    """
    Get queue declare info.

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

    Returns:
        Queue declare OK result from the broker.
    """
    timeout_ = timeout if timeout is not None else self._timeout

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

    return await channel.queue_declare(name, passive=True, timeout=timeout_)

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(_("declaring[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(_("deleting[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(_("deleting 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(
        _("binding[restore=%s] %s '%s' to exchange '%s' with routing_key '%s'"),
        restore,
        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(
        _("unbinding %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[str, Any] | 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[str, Any] | 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[str, Any] | 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] publishing[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,
                lambda msg: spec.callback(channel, msg),
                no_ack=spec.auto_ack,
                exclusive=spec.exclusive,
                arguments=spec.arguments.to_dict(),
                consumer_tag=spec.consumer_tag,
                timeout=timeout_,
            )
        ).consumer_tag,
    )

    logger.info(_("consuming[restore=%s] %s"), restore, 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 consuming %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[str, Any] | 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[str, Any] | 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[str, Any] | 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[str, Any] | 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[str, Any] | 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[str, Any] | 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[str, Any] | 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[str, Any] | 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

    @property
    def consumers(self) -> list[Consumer]:
        """Consumers attached to this queue."""
        return [c for c in self.ops.consumers if c.spec.queue == self.spec.name]

    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], Coroutine[Any, Any, Any]],
        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(),
        )
        return await self.ops.consume(spec, timeout=timeout, restore=restore)

    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 is not None:
            await self.ops.stop_consume(consumer_tag, timeout=timeout)
        else:
            for consumer in self.consumers:
                await self.ops.stop_consume(consumer.consumer_tag, timeout=timeout)

consumers property

consumers: list[Consumer]

Consumers attached to this queue.

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],
        Coroutine[Any, Any, Any],
    ],
    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], Coroutine[Any, Any, Any]]) –

  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], Coroutine[Any, Any, Any]],
    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(),
    )
    return await self.ops.consume(spec, timeout=timeout, restore=restore)

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 is not None:
        await self.ops.stop_consume(consumer_tag, timeout=timeout)
    else:
        for consumer in self.consumers:
            await self.ops.stop_consume(consumer.consumer_tag, timeout=timeout)

retry

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

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:

• Callable[[Callable[P, Awaitable]], Callable[P, Awaitable]] –

  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,
) -> Callable[[Callable[P, Awaitable]], Callable[P, Awaitable]]:
    """
    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