Skip to content

is_routing_exchange

faststream.rabbit.utils.is_routing_exchange #

is_routing_exchange(
    exchange: Optional[RabbitExchange],
) -> bool

Check if an exchange requires routing_key to deliver message.

Source code in faststream/rabbit/utils.py 
74
75
76
77
78
79
def is_routing_exchange(exchange: Optional["RabbitExchange"]) -> bool:
    """Check if an exchange requires routing_key to deliver message."""
    return not exchange or exchange.type in {
        ExchangeType.DIRECT.value,
        ExchangeType.TOPIC.value,
    }