Messages & Settlement

RabbitMessage

RabbitMessage

Rich incoming message with transport-aware settlement.

The message object wraps raw AMQP delivery data and provides: - Typed access to headers, properties, routing info - Sync and async ack/nack/reject methods - Idempotent settlement (double-ack is a no-op) - Topic wildcard path extraction

Source code in src/rabbitkit/core/message.py

class RabbitMessage:
    """Rich incoming message with transport-aware settlement.

    The message object wraps raw AMQP delivery data and provides:
    - Typed access to headers, properties, routing info
    - Sync and async ack/nack/reject methods
    - Idempotent settlement (double-ack is a no-op)
    - Topic wildcard path extraction
    """

    __slots__ = (
        "_ack_async_fn",
        "_ack_fn",
        "_disposition",
        "_nack_async_fn",
        "_nack_fn",
        "_reject_async_fn",
        "_reject_fn",
        "app_id",
        "body",
        "consumer_tag",
        "content_encoding",
        "content_type",
        "correlation_id",
        "delivery_tag",
        "exchange",
        "expiration",
        "headers",
        "message_id",
        "path",
        "priority",
        "raw_message",
        "redelivered",
        "reply_to",
        "routing_key",
        "timestamp",
        "type",
        "user_id",
    )

    def __init__(
        self,
        *,
        body: bytes,
        headers: dict[str, Any] | None = None,
        message_id: str | None = None,
        correlation_id: str | None = None,
        reply_to: str | None = None,
        content_type: str | None = None,
        content_encoding: str | None = None,
        timestamp: datetime | None = None,
        type: str | None = None,  # noqa: A002 — AMQP property name
        app_id: str | None = None,
        priority: int | None = None,
        expiration: str | None = None,
        user_id: str | None = None,
        routing_key: str = "",
        exchange: str = "",
        delivery_tag: int | None = None,
        redelivered: bool = False,
        consumer_tag: str | None = None,
        path: dict[str, str] | None = None,
        raw_message: Any = None,
    ) -> None:
        self.body = body
        self.headers: dict[str, Any] = headers or {}
        self.message_id = message_id
        self.correlation_id = correlation_id
        self.reply_to = reply_to
        self.content_type = content_type
        self.content_encoding = content_encoding
        self.timestamp = timestamp
        self.type = type
        self.app_id = app_id
        self.priority = priority
        self.expiration = expiration
        self.user_id = user_id
        self.routing_key = routing_key
        self.exchange = exchange
        self.delivery_tag = delivery_tag
        self.redelivered = redelivered
        self.consumer_tag = consumer_tag
        self.path: dict[str, str] = path or {}
        self.raw_message = raw_message

        # Transport-injected settlement functions (internal)
        self._ack_fn: Callable[[], None] | None = None
        self._ack_async_fn: Callable[[], Awaitable[None]] | None = None
        self._nack_fn: Callable[[bool], None] | None = None
        self._nack_async_fn: Callable[[bool], Awaitable[None]] | None = None
        self._reject_fn: Callable[[bool], None] | None = None
        self._reject_async_fn: Callable[[bool], Awaitable[None]] | None = None
        self._disposition: str = "pending"

    @property
    def is_settled(self) -> bool:
        """True if the message has been acked, nacked, or rejected."""
        return self._disposition != "pending"

    @property
    def disposition(self) -> str:
        """Final settlement state: "pending", "acked", "nacked", or "rejected" (M2)."""
        return self._disposition

    # ── Sync settlement ───────────────────────────────────────────────────

    def ack(self) -> None:
        """Synchronous ack. Raises SettlementError on async-only transport.

        Sets disposition only after the transport call succeeds, so a failed
        ack (channel closed, frame error) leaves the message unsettled and the
        exception propagates to the recovery loop instead of being swallowed.
        """
        if self._disposition != "pending":
            return  # idempotent guard
        if self._ack_fn is None:
            msg = "Cannot sync-ack an async transport message. Use await msg.ack_async()."
            raise SettlementError(msg)
        self._ack_fn()  # may raise — disposition stays "pending" on failure
        self._disposition = "acked"

    def nack(self, requeue: bool = True) -> None:
        """Synchronous nack. Raises SettlementError on async-only transport."""
        if self._disposition != "pending":
            return
        if self._nack_fn is None:
            msg = "Cannot sync-nack an async transport message. Use await msg.nack_async()."
            raise SettlementError(msg)
        self._nack_fn(requeue)
        self._disposition = "nacked"

    def reject(self, requeue: bool = False) -> None:
        """Synchronous reject. Raises SettlementError on async-only transport."""
        if self._disposition != "pending":
            return
        if self._reject_fn is None:
            msg = "Cannot sync-reject an async transport message. Use await msg.reject_async()."
            raise SettlementError(msg)
        self._reject_fn(requeue)
        self._disposition = "rejected"

    # ── Async settlement ──────────────────────────────────────────────────

    async def ack_async(self) -> None:
        """Async ack. Falls back to sync if async fn not set."""
        if self._disposition != "pending":
            return
        if self._ack_async_fn:
            await self._ack_async_fn()
        elif self._ack_fn:
            self._ack_fn()
        else:
            raise SettlementError("Cannot async-ack: no settlement fn set. Use msg.ack() on a sync transport.")
        self._disposition = "acked"

    async def nack_async(self, requeue: bool = True) -> None:
        """Async nack. Falls back to sync if async fn not set."""
        if self._disposition != "pending":
            return
        if self._nack_async_fn:
            await self._nack_async_fn(requeue)
        elif self._nack_fn:
            self._nack_fn(requeue)
        else:
            raise SettlementError("Cannot async-nack: no settlement fn set. Use msg.nack() on a sync transport.")
        self._disposition = "nacked"

    async def reject_async(self, requeue: bool = False) -> None:
        """Async reject. Falls back to sync if async fn not set."""
        if self._disposition != "pending":
            return
        if self._reject_async_fn:
            await self._reject_async_fn(requeue)
        elif self._reject_fn:
            self._reject_fn(requeue)
        else:
            raise SettlementError("Cannot async-reject: no settlement fn set. Use msg.reject() on a sync transport.")
        self._disposition = "rejected"

Attributes

is_settled: bool property

True if the message has been acked, nacked, or rejected.

disposition: str property

Final settlement state: "pending", "acked", "nacked", or "rejected" (M2).

Methods:

ack() -> None

Synchronous ack. Raises SettlementError on async-only transport.

Sets disposition only after the transport call succeeds, so a failed ack (channel closed, frame error) leaves the message unsettled and the exception propagates to the recovery loop instead of being swallowed.

Source code in src/rabbitkit/core/message.py

def ack(self) -> None:
    """Synchronous ack. Raises SettlementError on async-only transport.

    Sets disposition only after the transport call succeeds, so a failed
    ack (channel closed, frame error) leaves the message unsettled and the
    exception propagates to the recovery loop instead of being swallowed.
    """
    if self._disposition != "pending":
        return  # idempotent guard
    if self._ack_fn is None:
        msg = "Cannot sync-ack an async transport message. Use await msg.ack_async()."
        raise SettlementError(msg)
    self._ack_fn()  # may raise — disposition stays "pending" on failure
    self._disposition = "acked"

nack(requeue: bool = True) -> None

Synchronous nack. Raises SettlementError on async-only transport.

Source code in src/rabbitkit/core/message.py

def nack(self, requeue: bool = True) -> None:
    """Synchronous nack. Raises SettlementError on async-only transport."""
    if self._disposition != "pending":
        return
    if self._nack_fn is None:
        msg = "Cannot sync-nack an async transport message. Use await msg.nack_async()."
        raise SettlementError(msg)
    self._nack_fn(requeue)
    self._disposition = "nacked"

reject(requeue: bool = False) -> None

Synchronous reject. Raises SettlementError on async-only transport.

Source code in src/rabbitkit/core/message.py

def reject(self, requeue: bool = False) -> None:
    """Synchronous reject. Raises SettlementError on async-only transport."""
    if self._disposition != "pending":
        return
    if self._reject_fn is None:
        msg = "Cannot sync-reject an async transport message. Use await msg.reject_async()."
        raise SettlementError(msg)
    self._reject_fn(requeue)
    self._disposition = "rejected"

ack_async() -> None async

Async ack. Falls back to sync if async fn not set.

Source code in src/rabbitkit/core/message.py

async def ack_async(self) -> None:
    """Async ack. Falls back to sync if async fn not set."""
    if self._disposition != "pending":
        return
    if self._ack_async_fn:
        await self._ack_async_fn()
    elif self._ack_fn:
        self._ack_fn()
    else:
        raise SettlementError("Cannot async-ack: no settlement fn set. Use msg.ack() on a sync transport.")
    self._disposition = "acked"

nack_async(requeue: bool = True) -> None async

Async nack. Falls back to sync if async fn not set.

Source code in src/rabbitkit/core/message.py

async def nack_async(self, requeue: bool = True) -> None:
    """Async nack. Falls back to sync if async fn not set."""
    if self._disposition != "pending":
        return
    if self._nack_async_fn:
        await self._nack_async_fn(requeue)
    elif self._nack_fn:
        self._nack_fn(requeue)
    else:
        raise SettlementError("Cannot async-nack: no settlement fn set. Use msg.nack() on a sync transport.")
    self._disposition = "nacked"

reject_async(requeue: bool = False) -> None async

Async reject. Falls back to sync if async fn not set.

Source code in src/rabbitkit/core/message.py

async def reject_async(self, requeue: bool = False) -> None:
    """Async reject. Falls back to sync if async fn not set."""
    if self._disposition != "pending":
        return
    if self._reject_async_fn:
        await self._reject_async_fn(requeue)
    elif self._reject_fn:
        self._reject_fn(requeue)
    else:
        raise SettlementError("Cannot async-reject: no settlement fn set. Use msg.reject() on a sync transport.")
    self._disposition = "rejected"

MessageEnvelope

MessageEnvelope dataclass

Outgoing message envelope.

NOTE: AMQP header values are limited to: str, int, float, bool, bytes, datetime, Decimal, list/dict of these, or None. Arbitrary Python objects (sets, custom classes) will raise at publish time. Transport validates header values before sending.

Source code in src/rabbitkit/core/types.py

@dataclass(frozen=True, slots=True)
class MessageEnvelope:
    """Outgoing message envelope.

    NOTE: AMQP header values are limited to:
    str, int, float, bool, bytes, datetime, Decimal, list/dict of these, or None.
    Arbitrary Python objects (sets, custom classes) will raise at publish time.
    Transport validates header values before sending.
    """

    routing_key: str
    body: bytes
    exchange: str = ""
    headers: dict[str, Any] = field(default_factory=dict)
    message_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    correlation_id: str | None = None
    reply_to: str | None = None
    timestamp: datetime | None = None
    content_type: str = "application/json"
    content_encoding: str | None = None
    expiration: str | None = None
    priority: int | None = None
    mandatory: bool = False
    delivery_mode: int = 2  # 1=transient, 2=persistent
    type: str | None = None
    user_id: str | None = None
    app_id: str | None = None

    def __post_init__(self) -> None:
        # Catches an oversized routing_key/exchange at construction time --
        # the same choke point every publish (broker.publish, retry
        # republish, DLQ replay, batch) goes through -- instead of an
        # opaque broker connection error later.
        validate_amqp_shortstr("routing_key", self.routing_key)
        validate_amqp_shortstr("exchange", self.exchange)

AckPolicy

AckPolicy

Bases: str, Enum

Message acknowledgement policies.

See Contract 1 in the plan for exact semantics: - AUTO: success→ack, exception→classify→nack/reject - MANUAL: handler owns ack/nack/reject entirely - NACK_ON_ERROR: success→ack, exception→nack(requeue=False) - ACK_FIRST: ack BEFORE handler runs (at-most-once)

Source code in src/rabbitkit/core/types.py

class AckPolicy(str, Enum):
    """Message acknowledgement policies.

    See Contract 1 in the plan for exact semantics:
    - AUTO: success→ack, exception→classify→nack/reject
    - MANUAL: handler owns ack/nack/reject entirely
    - NACK_ON_ERROR: success→ack, exception→nack(requeue=False)
    - ACK_FIRST: ack BEFORE handler runs (at-most-once)
    """

    AUTO = "auto"
    MANUAL = "manual"
    NACK_ON_ERROR = "nack_on_error"
    ACK_FIRST = "ack_first"

AckStrategy

AckStrategy

Bases: Protocol

Settlement strategy for an AckPolicy.

Each strategy owns the success-path ack and the error-path settlement. Handler-raised AckMessage / NackMessage / RejectMessage are NOT policy-driven and stay in the pipeline.

See Contract 1 in the plan for per-policy semantics.

Source code in src/rabbitkit/core/types.py

@runtime_checkable
class AckStrategy(Protocol):
    """Settlement strategy for an ``AckPolicy``.

    Each strategy owns the success-path ack and the error-path settlement.
    Handler-raised ``AckMessage`` / ``NackMessage`` / ``RejectMessage`` are
    NOT policy-driven and stay in the pipeline.

    See Contract 1 in the plan for per-policy semantics.
    """

    @property
    def acks_first(self) -> bool:
        """True when the message is acked BEFORE the handler runs (ACK_FIRST)."""
        ...

    def on_success(self, msg: RabbitMessage) -> None:
        """Settle the message after a successful handler invocation."""
        ...

    def on_error(self, msg: RabbitMessage, exc: Exception) -> None:
        """Settle the message after an unhandled handler exception."""
        ...

Attributes

acks_first: bool property

True when the message is acked BEFORE the handler runs (ACK_FIRST).

Methods:

on_success(msg: RabbitMessage) -> None

Settle the message after a successful handler invocation.

Source code in src/rabbitkit/core/types.py

def on_success(self, msg: RabbitMessage) -> None:
    """Settle the message after a successful handler invocation."""
    ...

on_error(msg: RabbitMessage, exc: Exception) -> None

Settle the message after an unhandled handler exception.

Source code in src/rabbitkit/core/types.py

def on_error(self, msg: RabbitMessage, exc: Exception) -> None:
    """Settle the message after an unhandled handler exception."""
    ...

PublishOutcome / PublishStatus

PublishOutcome dataclass

Result of a publish operation.

Source code in src/rabbitkit/core/types.py

@dataclass(frozen=True, slots=True)
class PublishOutcome:
    """Result of a publish operation."""

    status: PublishStatus
    delivery_tag: int | None = None
    exchange: str = ""
    routing_key: str = ""
    error: BaseException | None = None
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))

    @property
    def ok(self) -> bool:
        """True if the publish did not fail -- CONFIRMED (broker
        acknowledged it) or SENT (fire-and-forget, confirm_delivery=False --
        written to the socket but never broker-confirmed).

        M4: if you specifically need to know the broker actually confirmed
        the message (e.g. before treating a republish as durable enough to
        settle/discard the original), check ``status ==
        PublishStatus.CONFIRMED`` directly -- ``.ok`` alone can't
        distinguish "confirmed" from "sent, unconfirmed."
        """
        return self.status in (PublishStatus.CONFIRMED, PublishStatus.SENT)

    def raise_for_status(self) -> PublishOutcome:
        """Raise ``PublishError`` if the publish failed; else return self (M1).

        ``broker.publish()`` never raises — it returns this outcome so a
        failed publish (NACKED / TIMEOUT / RETURNED / ERROR) can't be lost by
        code that simply ignores the return value. Callers who prefer
        exceptions opt in::

            broker.publish(envelope).raise_for_status()

        Chains so ``outcome = broker.publish(...).raise_for_status()`` works.
        """
        if not self.ok:
            from rabbitkit.core.errors import PublishError

            raise PublishError(self)
        return self

    @property
    def classification(self) -> ClassifiedError | None:
        """Best-effort severity/reason for ``self.error``, via the existing
        ``classify_error()`` (production-hardening item 5).

        Deliberately NOT a new ``PublishStatus`` enum member: growing that
        small, closed "what happened" set (CONFIRMED/SENT/NACKED/TIMEOUT/
        RETURNED/ERROR) risks breaking exhaustive match/if-chains callers
        may have written against it, for a "why" signal better served by
        reusing the classification machinery already wired into the
        consume/retry path (``core/errors.py``) rather than duplicating it.
        A lazy import avoids a module-level cycle (``core/errors.py``
        already imports from this module) — same pattern as
        :meth:`raise_for_status`.

        ``None`` when there's nothing to classify: a non-ERROR status, or
        an ERROR outcome that (unusually) captured no exception.
        """
        if self.status is not PublishStatus.ERROR or self.error is None:
            return None
        from rabbitkit.core.errors import classify_error

        return classify_error(self.error)

Attributes

ok: bool property

True if the publish did not fail -- CONFIRMED (broker acknowledged it) or SENT (fire-and-forget, confirm_delivery=False -- written to the socket but never broker-confirmed).

M4: if you specifically need to know the broker actually confirmed the message (e.g. before treating a republish as durable enough to settle/discard the original), check status == PublishStatus.CONFIRMED directly -- .ok alone can't distinguish "confirmed" from "sent, unconfirmed."

classification: ClassifiedError | None property

Best-effort severity/reason for self.error, via the existing classify_error() (production-hardening item 5).

Deliberately NOT a new PublishStatus enum member: growing that small, closed "what happened" set (CONFIRMED/SENT/NACKED/TIMEOUT/ RETURNED/ERROR) risks breaking exhaustive match/if-chains callers may have written against it, for a "why" signal better served by reusing the classification machinery already wired into the consume/retry path (core/errors.py) rather than duplicating it. A lazy import avoids a module-level cycle (core/errors.py already imports from this module) — same pattern as :meth:raise_for_status.

None when there's nothing to classify: a non-ERROR status, or an ERROR outcome that (unusually) captured no exception.

Methods:

raise_for_status() -> PublishOutcome

Raise PublishError if the publish failed; else return self (M1).

broker.publish() never raises — it returns this outcome so a failed publish (NACKED / TIMEOUT / RETURNED / ERROR) can't be lost by code that simply ignores the return value. Callers who prefer exceptions opt in::

broker.publish(envelope).raise_for_status()

Chains so outcome = broker.publish(...).raise_for_status() works.

Source code in src/rabbitkit/core/types.py

def raise_for_status(self) -> PublishOutcome:
    """Raise ``PublishError`` if the publish failed; else return self (M1).

    ``broker.publish()`` never raises — it returns this outcome so a
    failed publish (NACKED / TIMEOUT / RETURNED / ERROR) can't be lost by
    code that simply ignores the return value. Callers who prefer
    exceptions opt in::

        broker.publish(envelope).raise_for_status()

    Chains so ``outcome = broker.publish(...).raise_for_status()`` works.
    """
    if not self.ok:
        from rabbitkit.core.errors import PublishError

        raise PublishError(self)
    return self

PublishStatus

Bases: str, Enum

Result status of a publish operation.

Source code in src/rabbitkit/core/types.py

class PublishStatus(str, Enum):
    """Result status of a publish operation."""

    CONFIRMED = "confirmed"
    #: M4: fire-and-forget publish (PublisherConfig.confirm_delivery=False)
    #: -- written to the socket, but the broker never acknowledged it.
    #: Distinct from CONFIRMED so code that specifically needs a real
    #: broker ack (e.g. deciding whether it's safe to ack/discard a source
    #: message after republishing it, as retry/result publishing do) can
    #: tell the two apart via ``.status`` instead of being told "confirmed"
    #: when nothing was actually confirmed.
    SENT = "sent"
    NACKED = "nacked"
    TIMEOUT = "timeout"
    RETURNED = "returned"
    ERROR = "error"

Settlement exceptions

AckMessage

Bases: Exception

Raise from handler to ack the message.

Source code in src/rabbitkit/core/message.py

class AckMessage(Exception):
    """Raise from handler to ack the message."""

NackMessage

Bases: Exception

Raise from handler to nack the message.

Source code in src/rabbitkit/core/message.py

class NackMessage(Exception):
    """Raise from handler to nack the message."""

    def __init__(self, requeue: bool = True) -> None:
        super().__init__()
        self.requeue = requeue

RejectMessage

Bases: Exception

Raise from handler to reject the message.

Source code in src/rabbitkit/core/message.py

class RejectMessage(Exception):
    """Raise from handler to reject the message."""

    def __init__(self, requeue: bool = False) -> None:
        super().__init__()
        self.requeue = requeue