Skip to content

is_routing_exchange

faststream.rabbit.utils.is_routing_exchange #

is_routing_exchange(exchange)

Check if an exchange requires routing_key to deliver message.

Source code in faststream/rabbit/utils.py 
50
51
52
53
54
55
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,
    )