Management API

RabbitManagementClient

RabbitManagementClient

HTTP client for the RabbitMQ Management API.

Source code in src/rabbitkit/management.py

class RabbitManagementClient:
    """HTTP client for the RabbitMQ Management API."""

    def __init__(self, config: ManagementConfig | None = None) -> None:
        self._config = config or ManagementConfig()
        credentials = f"{self._config.username}:{self._config.password}"
        self._auth_header = "Basic " + base64.b64encode(credentials.encode()).decode()
        # No-redirect opener: 3xx raise HTTPError instead of being followed.
        self._opener = urllib.request.build_opener(_NoRedirect)
        # Long-lived aiohttp session reused across *_async requests (L-6).
        # Lazily created on the first async request; closed via aclose().
        self._aiohttp_session: Any = None

    def _request(self, method: str, path: str, body: bytes | None = None) -> Any:
        url = f"{self._config.url}/api{path}"
        req = urllib.request.Request(url, method=method, data=body)  # noqa: S310
        req.add_header("Authorization", self._auth_header)
        req.add_header("Content-Type", "application/json")
        try:
            resp = self._opener.open(req, timeout=self._config.timeout)
        except urllib.error.HTTPError as exc:
            # With _NoRedirect installed, 3xx surface here as HTTPError.
            if 300 <= exc.code < 400:
                raise ValueError(f"Unexpected redirect ({exc.code}) to {exc.headers.get('Location')}") from exc
            raise
        with resp:
            if resp.status == 204:
                # Drain + discard any (usually empty) body.
                return None
            data = self._read_capped(resp)
            if not data:
                # e.g. 201 Created from PUT /queues, /parameters, /bindings — empty body.
                return None
            return json.loads(data.decode())

    def _read_capped(self, resp: Any) -> bytes:
        """Read the response body in chunks, raising if it exceeds the cap."""
        total = 0
        chunks: list[bytes] = []
        while True:
            chunk = resp.read(_READ_CHUNK)
            if not chunk:
                break
            total += len(chunk)
            if total > _MAX_RESPONSE_BYTES:
                raise ValueError(f"Management API response exceeded {_MAX_RESPONSE_BYTES} bytes")
            chunks.append(chunk)
        return b"".join(chunks)

    # Queue operations
    def list_queues(self, vhost: str = "/") -> list[QueueInfo]:
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        return cast("list[QueueInfo]", self._request("GET", f"/queues/{vhost_encoded}"))

    def get_queue(self, name: str, vhost: str = "/") -> QueueInfo:
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        name_encoded = urllib.parse.quote(name, safe="")
        return cast("QueueInfo", self._request("GET", f"/queues/{vhost_encoded}/{name_encoded}"))

    def purge_queue(self, name: str, vhost: str = "/") -> None:
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        name_encoded = urllib.parse.quote(name, safe="")
        self._request("DELETE", f"/queues/{vhost_encoded}/{name_encoded}/contents")

    def delete_queue(self, name: str, vhost: str = "/") -> None:
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        name_encoded = urllib.parse.quote(name, safe="")
        self._request("DELETE", f"/queues/{vhost_encoded}/{name_encoded}")

    # Exchange operations
    def list_exchanges(self, vhost: str = "/") -> list[ExchangeInfo]:
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        return cast("list[ExchangeInfo]", self._request("GET", f"/exchanges/{vhost_encoded}"))

    def get_exchange(self, name: str, vhost: str = "/") -> ExchangeInfo:
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        name_encoded = urllib.parse.quote(name, safe="")
        return cast("ExchangeInfo", self._request("GET", f"/exchanges/{vhost_encoded}/{name_encoded}"))

    # Queue declaration / bindings (used by `rabbitkit topology migrate`)
    def declare_queue(
        self,
        name: str,
        vhost: str = "/",
        durable: bool = True,
        arguments: dict[str, Any] | None = None,
    ) -> None:
        """Declare a queue via ``PUT /api/queues/{vhost}/{name}``.

        ``arguments`` are the queue's x-arguments (e.g. ``{"x-queue-type": "quorum"}``).
        """
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        name_encoded = urllib.parse.quote(name, safe="")
        body = json.dumps({"durable": durable, "auto_delete": False, "arguments": arguments or {}}).encode()
        self._request("PUT", f"/queues/{vhost_encoded}/{name_encoded}", body)

    def get_queue_bindings(self, queue: str, vhost: str = "/") -> list[dict[str, Any]]:
        """List all bindings of a queue via ``GET /api/queues/{vhost}/{queue}/bindings``."""
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        queue_encoded = urllib.parse.quote(queue, safe="")
        return cast(
            "list[dict[str, Any]]",
            self._request("GET", f"/queues/{vhost_encoded}/{queue_encoded}/bindings"),
        )

    def bind_queue(
        self,
        queue: str,
        exchange: str,
        routing_key: str = "",
        vhost: str = "/",
        arguments: dict[str, Any] | None = None,
    ) -> None:
        """Bind a queue to an exchange via ``POST /api/bindings/{vhost}/e/{exchange}/q/{queue}``."""
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        exchange_encoded = urllib.parse.quote(exchange, safe="")
        queue_encoded = urllib.parse.quote(queue, safe="")
        body = json.dumps({"routing_key": routing_key, "arguments": arguments or {}}).encode()
        self._request("POST", f"/bindings/{vhost_encoded}/e/{exchange_encoded}/q/{queue_encoded}", body)

    # Runtime parameters (dynamic shovels, federation upstreams, ...)
    def put_parameter(self, component: str, vhost: str, name: str, value: dict[str, Any]) -> None:
        """Create/update a runtime parameter via ``PUT /api/parameters/{component}/{vhost}/{name}``.

        For a dynamic shovel: ``put_parameter("shovel", "/", "my-shovel", {"value": {...}})``.
        """
        component_encoded = urllib.parse.quote(component, safe="")
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        name_encoded = urllib.parse.quote(name, safe="")
        body = json.dumps(value).encode()
        self._request("PUT", f"/parameters/{component_encoded}/{vhost_encoded}/{name_encoded}", body)

    def delete_parameter(self, component: str, vhost: str, name: str) -> None:
        """Delete a runtime parameter via ``DELETE /api/parameters/{component}/{vhost}/{name}``."""
        component_encoded = urllib.parse.quote(component, safe="")
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        name_encoded = urllib.parse.quote(name, safe="")
        self._request("DELETE", f"/parameters/{component_encoded}/{vhost_encoded}/{name_encoded}")

    def list_shovel_statuses(self) -> list[dict[str, Any]]:
        """List shovel statuses via ``GET /api/shovels``.

        Raises (HTTP 404) when the ``rabbitmq_shovel`` plugin is not enabled.
        """
        return cast("list[dict[str, Any]]", self._request("GET", "/shovels"))

    # Connection/Channel
    def list_connections(self) -> list[ConnectionInfo]:
        return cast("list[ConnectionInfo]", self._request("GET", "/connections"))

    def list_channels(self) -> list[ChannelInfo]:
        return cast("list[ChannelInfo]", self._request("GET", "/channels"))

    # Overview
    def overview(self) -> OverviewInfo:
        return cast("OverviewInfo", self._request("GET", "/overview"))

    def health_check(self) -> bool:
        try:
            result = self._request("GET", "/healthchecks/node")
            return bool(result.get("status") == "ok")
        except Exception:
            return False

    # Async variants
    async def list_queues_async(self, vhost: str = "/") -> list[QueueInfo]:
        vhost_encoded = urllib.parse.quote(vhost, safe="")
        return cast("list[QueueInfo]", await self._request_async("GET", f"/queues/{vhost_encoded}"))

    async def get_queue_async(self, name: str, vhost: str = "/") -> QueueInfo:
        v = urllib.parse.quote(vhost, safe="")
        n = urllib.parse.quote(name, safe="")
        return cast("QueueInfo", await self._request_async("GET", f"/queues/{v}/{n}"))

    async def overview_async(self) -> OverviewInfo:
        return cast("OverviewInfo", await self._request_async("GET", "/overview"))

    async def health_check_async(self) -> bool:
        try:
            result = await self._request_async("GET", "/healthchecks/node")
            return bool(result.get("status") == "ok")
        except Exception:
            return False

    async def _read_capped_async(self, resp: Any) -> bytes:
        """Read the async response body in chunks, raising if it exceeds the cap."""
        total = 0
        chunks: list[bytes] = []
        while True:
            chunk = await resp.content.read(_READ_CHUNK)
            if not chunk:
                break
            total += len(chunk)
            if total > _MAX_RESPONSE_BYTES:
                raise ValueError(f"Management API response exceeded {_MAX_RESPONSE_BYTES} bytes")
            chunks.append(chunk)
        return b"".join(chunks)

    async def _get_session(self) -> Any:
        """Lazily create and reuse a single aiohttp.ClientSession (L-6)."""
        try:
            import aiohttp
        except ImportError:
            raise ImportError(
                "Async management API requires aiohttp. Install with: pip install rabbitkit[management]"
            ) from None
        if self._aiohttp_session is None:
            self._aiohttp_session = aiohttp.ClientSession()
        return self._aiohttp_session

    async def aclose(self) -> None:
        """Close the long-lived aiohttp session if one was created.

        ``*_async`` methods reuse a single ``aiohttp.ClientSession`` for
        connection pooling. Call ``aclose()`` at shutdown to release the
        underlying connector's sockets; otherwise the session is cleaned up
        when the event loop closes.
        """
        if self._aiohttp_session is not None:
            await self._aiohttp_session.close()
            self._aiohttp_session = None

    async def _request_async(self, method: str, path: str, body: bytes | None = None) -> Any:
        try:
            import aiohttp
        except ImportError:
            raise ImportError(
                "Async management API requires aiohttp. Install with: pip install rabbitkit[management]"
            ) from None
        session = await self._get_session()

        url = f"{self._config.url}/api{path}"
        headers = {"Authorization": self._auth_header, "Content-Type": "application/json"}
        async with session.request(
            method,
            url,
            headers=headers,
            data=body,
            timeout=aiohttp.ClientTimeout(total=self._config.timeout),
            allow_redirects=False,  # never follow — surface 3xx as an error (SSRF guard)
        ) as resp:
            if 300 <= resp.status < 400:
                raise ValueError(f"Unexpected redirect ({resp.status}) to {resp.headers.get('Location')}")
            if resp.status == 204:
                return None
            if not (200 <= resp.status < 300):
                resp.raise_for_status()  # match the sync path: raise on non-2xx
            return json.loads((await self._read_capped_async(resp)).decode())

Methods:

declare_queue(name: str, vhost: str = '/', durable: bool = True, arguments: dict[str, Any] | None = None) -> None

Declare a queue via PUT /api/queues/{vhost}/{name}.

arguments are the queue's x-arguments (e.g. {"x-queue-type": "quorum"}).

Source code in src/rabbitkit/management.py

def declare_queue(
    self,
    name: str,
    vhost: str = "/",
    durable: bool = True,
    arguments: dict[str, Any] | None = None,
) -> None:
    """Declare a queue via ``PUT /api/queues/{vhost}/{name}``.

    ``arguments`` are the queue's x-arguments (e.g. ``{"x-queue-type": "quorum"}``).
    """
    vhost_encoded = urllib.parse.quote(vhost, safe="")
    name_encoded = urllib.parse.quote(name, safe="")
    body = json.dumps({"durable": durable, "auto_delete": False, "arguments": arguments or {}}).encode()
    self._request("PUT", f"/queues/{vhost_encoded}/{name_encoded}", body)

get_queue_bindings(queue: str, vhost: str = '/') -> list[dict[str, Any]]

List all bindings of a queue via GET /api/queues/{vhost}/{queue}/bindings.

Source code in src/rabbitkit/management.py

def get_queue_bindings(self, queue: str, vhost: str = "/") -> list[dict[str, Any]]:
    """List all bindings of a queue via ``GET /api/queues/{vhost}/{queue}/bindings``."""
    vhost_encoded = urllib.parse.quote(vhost, safe="")
    queue_encoded = urllib.parse.quote(queue, safe="")
    return cast(
        "list[dict[str, Any]]",
        self._request("GET", f"/queues/{vhost_encoded}/{queue_encoded}/bindings"),
    )

bind_queue(queue: str, exchange: str, routing_key: str = '', vhost: str = '/', arguments: dict[str, Any] | None = None) -> None

Bind a queue to an exchange via POST /api/bindings/{vhost}/e/{exchange}/q/{queue}.

Source code in src/rabbitkit/management.py

def bind_queue(
    self,
    queue: str,
    exchange: str,
    routing_key: str = "",
    vhost: str = "/",
    arguments: dict[str, Any] | None = None,
) -> None:
    """Bind a queue to an exchange via ``POST /api/bindings/{vhost}/e/{exchange}/q/{queue}``."""
    vhost_encoded = urllib.parse.quote(vhost, safe="")
    exchange_encoded = urllib.parse.quote(exchange, safe="")
    queue_encoded = urllib.parse.quote(queue, safe="")
    body = json.dumps({"routing_key": routing_key, "arguments": arguments or {}}).encode()
    self._request("POST", f"/bindings/{vhost_encoded}/e/{exchange_encoded}/q/{queue_encoded}", body)

put_parameter(component: str, vhost: str, name: str, value: dict[str, Any]) -> None

Create/update a runtime parameter via PUT /api/parameters/{component}/{vhost}/{name}.

For a dynamic shovel: put_parameter("shovel", "/", "my-shovel", {"value": {...}}).

Source code in src/rabbitkit/management.py

def put_parameter(self, component: str, vhost: str, name: str, value: dict[str, Any]) -> None:
    """Create/update a runtime parameter via ``PUT /api/parameters/{component}/{vhost}/{name}``.

    For a dynamic shovel: ``put_parameter("shovel", "/", "my-shovel", {"value": {...}})``.
    """
    component_encoded = urllib.parse.quote(component, safe="")
    vhost_encoded = urllib.parse.quote(vhost, safe="")
    name_encoded = urllib.parse.quote(name, safe="")
    body = json.dumps(value).encode()
    self._request("PUT", f"/parameters/{component_encoded}/{vhost_encoded}/{name_encoded}", body)

delete_parameter(component: str, vhost: str, name: str) -> None

Delete a runtime parameter via DELETE /api/parameters/{component}/{vhost}/{name}.

Source code in src/rabbitkit/management.py

def delete_parameter(self, component: str, vhost: str, name: str) -> None:
    """Delete a runtime parameter via ``DELETE /api/parameters/{component}/{vhost}/{name}``."""
    component_encoded = urllib.parse.quote(component, safe="")
    vhost_encoded = urllib.parse.quote(vhost, safe="")
    name_encoded = urllib.parse.quote(name, safe="")
    self._request("DELETE", f"/parameters/{component_encoded}/{vhost_encoded}/{name_encoded}")

list_shovel_statuses() -> list[dict[str, Any]]

List shovel statuses via GET /api/shovels.

Raises (HTTP 404) when the rabbitmq_shovel plugin is not enabled.

Source code in src/rabbitkit/management.py

def list_shovel_statuses(self) -> list[dict[str, Any]]:
    """List shovel statuses via ``GET /api/shovels``.

    Raises (HTTP 404) when the ``rabbitmq_shovel`` plugin is not enabled.
    """
    return cast("list[dict[str, Any]]", self._request("GET", "/shovels"))

aclose() -> None async

Close the long-lived aiohttp session if one was created.

*_async methods reuse a single aiohttp.ClientSession for connection pooling. Call aclose() at shutdown to release the underlying connector's sockets; otherwise the session is cleaned up when the event loop closes.

Source code in src/rabbitkit/management.py

async def aclose(self) -> None:
    """Close the long-lived aiohttp session if one was created.

    ``*_async`` methods reuse a single ``aiohttp.ClientSession`` for
    connection pooling. Call ``aclose()`` at shutdown to release the
    underlying connector's sockets; otherwise the session is cleaned up
    when the event loop closes.
    """
    if self._aiohttp_session is not None:
        await self._aiohttp_session.close()
        self._aiohttp_session = None

ManagementConfig

ManagementConfig dataclass

RabbitMQ Management API configuration.

.. warning:: The default URL uses http:// with guest/guest credentials — this is intended only for local development. In production use https:// with non-default credentials. Schemes other than http/https are rejected to prevent SSRF via crafted URLs.

Source code in src/rabbitkit/management.py

@dataclass(frozen=True, slots=True)
class ManagementConfig:
    """RabbitMQ Management API configuration.

    .. warning::
        The default URL uses ``http://`` with ``guest``/``guest`` credentials —
        this is intended **only for local development**. In production use
        ``https://`` with non-default credentials. Schemes other than
        ``http``/``https`` are rejected to prevent SSRF via crafted URLs.
    """

    url: str = "http://localhost:15672"
    username: str = "guest"
    password: str = "guest"
    timeout: float = 10.0

    def __repr__(self) -> str:
        """L2: mask the password — the default dataclass repr would leak it
        in plaintext into any log line or traceback that reprs this object."""
        from rabbitkit.core.config import _masked_repr

        return _masked_repr(self)

    def __post_init__(self) -> None:
        scheme = urllib.parse.urlparse(self.url).scheme.lower()
        if scheme not in {"http", "https"}:
            raise ValueError(f"Unsupported management URL scheme {scheme!r}; only 'http' and 'https' are allowed.")
        # Mirror ConnectionConfig.__post_init__: warn when the default 'guest'
        # credentials are used against a non-local host (dev convenience, but
        # flag the production misconfiguration once at construction).
        hostname = urllib.parse.urlparse(self.url).hostname
        if self.username == "guest" and hostname not in {"localhost", "127.0.0.1", "::1", None}:
            warnings.warn(
                "ManagementConfig uses default 'guest' credentials against non-local "
                f"host {hostname!r}; set explicit username/password for production.",
                UserWarning,
                stacklevel=2,
            )

Methods:

__repr__() -> str

L2: mask the password — the default dataclass repr would leak it in plaintext into any log line or traceback that reprs this object.

Source code in src/rabbitkit/management.py

def __repr__(self) -> str:
    """L2: mask the password — the default dataclass repr would leak it
    in plaintext into any log line or traceback that reprs this object."""
    from rabbitkit.core.config import _masked_repr

    return _masked_repr(self)

QueueInfo (TypedDict)

QueueInfo

Bases: TypedDict

Typed view of a RabbitMQ Management API queue response.

The Management API returns many fields; this captures the most common ones. total=False means every key is optional — the API may omit fields depending on the endpoint and RabbitMQ version.

Source code in src/rabbitkit/management.py

class QueueInfo(TypedDict, total=False):
    """Typed view of a RabbitMQ Management API queue response.

    The Management API returns many fields; this captures the most common
    ones. ``total=False`` means every key is optional — the API may omit
    fields depending on the endpoint and RabbitMQ version.
    """

    name: str
    vhost: str
    type: str
    durable: bool
    auto_delete: bool
    messages: int
    messages_ready: int
    messages_unacknowledged: int
    consumers: int
    state: str