RabbitExchange

faststream.rabbit.schemas.RabbitExchange

RabbitExchange(
    name="",
    type=DIRECT,
    durable=False,
    auto_delete=False,
    declare=True,
    arguments=None,
    timeout=None,
    robust=True,
    bind_to=None,
    bind_arguments=None,
    routing_key="",
)

Bases: NameRequired

A class to represent a RabbitMQ exchange.

Initialize a RabbitExchange object.

PARAMETER	DESCRIPTION
name	RabbitMQ exchange name. TYPE: str DEFAULT: ''
type	RabbitMQ exchange type. TYPE: ExchangeType DEFAULT: DIRECT
durable	Whether the object is durable. TYPE: bool DEFAULT: False
auto_delete	The exchange will be deleted after connection closed. TYPE: bool DEFAULT: False
declare	Whether to exchange automatically or just connect to it. TYPE: bool DEFAULT: True
arguments	Exchange declarationg arguments. TYPE: dict[str, Any] | None DEFAULT: None
timeout	Send confirmation time from RabbitMQ. TYPE: TimeoutType DEFAULT: None
robust	Whether to declare exchange object as restorable. TYPE: bool DEFAULT: True
bind_to	Another RabbitExchange object to bind the current one to. TYPE: Optional[RabbitExchange] DEFAULT: None
bind_arguments	Exchange-exchange binding options. TYPE: dict[str, Any] | None DEFAULT: None
routing_key	Explicit binding routing key. TYPE: str DEFAULT: ''

Source code in faststream/rabbit/schemas/exchange.py

def __init__(
    self,
    name: str = "",
    type: ExchangeType = ExchangeType.DIRECT,
    durable: bool = False,
    auto_delete: bool = False,
    # custom
    declare: bool = True,
    arguments: dict[str, Any] | None = None,
    timeout: "TimeoutType" = None,
    robust: bool = True,
    bind_to: Optional["RabbitExchange"] = None,
    bind_arguments: dict[str, Any] | None = None,
    routing_key: str = "",
) -> None:
    """Initialize a RabbitExchange object.

    Args:
        name: RabbitMQ exchange name.
        type: RabbitMQ exchange type.
        durable: Whether the object is durable.
        auto_delete: The exchange will be deleted after connection closed.
        declare: Whether to exchange automatically or just connect to it.
        arguments: Exchange declarationg arguments.
        timeout: Send confirmation time from RabbitMQ.
        robust: Whether to declare exchange object as restorable.
        bind_to: Another `RabbitExchange` object to bind the current one to.
        bind_arguments: Exchange-exchange binding options.
        routing_key: Explicit binding routing key.
    """
    if routing_key and bind_to is None:  # pragma: no cover
        warnings.warn(
            (
                "\nRabbitExchange `routing_key` is using to bind exchange to another one."
                "\nIt can be used only with the `bind_to` argument, please setup it too."
            ),
            category=RuntimeWarning,
            stacklevel=1,
        )

    super().__init__(name)

    self.type = type
    self.durable = durable
    self.auto_delete = auto_delete
    self.robust = robust
    self.timeout = timeout
    self.arguments = arguments
    self.declare = declare
    self.bind_to = bind_to
    self.bind_arguments = bind_arguments
    self.routing_key = routing_key

name instance-attribute

name = name

type instance-attribute

type = type

durable instance-attribute

durable = durable

auto_delete instance-attribute

auto_delete = auto_delete

robust instance-attribute

robust = robust

timeout instance-attribute

timeout = timeout

arguments instance-attribute

arguments = arguments

declare instance-attribute

declare = declare

bind_to instance-attribute

bind_to = bind_to

bind_arguments instance-attribute

bind_arguments = bind_arguments

routing_key instance-attribute

routing_key = routing_key

routing

routing()

Return real routing_key of object.

Source code in faststream/rabbit/schemas/exchange.py

def routing(self) -> str:
    """Return real routing_key of object."""
    return self.routing_key or self.name

validate classmethod

validate(value, **kwargs)

Source code in faststream/rabbit/schemas/exchange.py

@override
@classmethod
def validate(  # type: ignore[override]
    cls,
    value: Union[str, "RabbitExchange", None],
    **kwargs: Any,
) -> "RabbitExchange":
    exch = super().validate(value, **kwargs)
    if exch is None:
        exch = RabbitExchange()
    return exch