Skip to content

RabbitExchange

faststream.rabbit.RabbitExchange #

RabbitExchange(
    name: str = "",
    type: ExchangeType = DIRECT,
    durable: bool = False,
    auto_delete: bool = False,
    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 = "",
)

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 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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() -> str

Return real routing_key of object.

Source code in faststream/rabbit/schemas/exchange.py 
49
50
51
def routing(self) -> str:
    """Return real routing_key of object."""
    return self.routing_key or self.name

validate classmethod #

validate(
    value: Union[str, RabbitExchange, None], **kwargs: Any
) -> RabbitExchange
Source code in faststream/rabbit/schemas/exchange.py 
106
107
108
109
110
111
112
113
114
115
116
@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