Skip to content

RabbitExchange

faststream.rabbit.shared.schemas.RabbitExchange #

RabbitExchange(
    name: str,
    type: ExchangeType = ExchangeType.DIRECT,
    durable: bool = False,
    auto_delete: bool = False,
    internal: bool = False,
    passive: bool = False,
    arguments: Optional[AnyDict] = None,
    timeout: TimeoutType = None,
    robust: bool = True,
    bind_to: Optional[RabbitExchange] = None,
    bind_arguments: Optional[AnyDict] = None,
    routing_key: str = "",
)

Bases: NameRequired

A class to represent a RabbitMQ exchange.

METHOD DESCRIPTION
__hash__

returns the hash value of the exchange

__init__

initializes the RabbitExchange object

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Initialize a RabbitExchange object.

PARAMETER DESCRIPTION
name

Name of the exchange.

TYPE: str

type

Type of the exchange. Defaults to ExchangeType.DIRECT.

TYPE: ExchangeType DEFAULT: DIRECT

durable

Whether the exchange should survive broker restarts. Defaults to False.

TYPE: bool DEFAULT: False

auto_delete

Whether the exchange should be deleted when no longer in use. Defaults to False.

TYPE: bool DEFAULT: False

internal

Whether the exchange is used for internal purposes and should not be published to directly. Defaults to False.

TYPE: bool DEFAULT: False

passive

Whether to check if the exchange exists before creating it. Defaults to False.

TYPE: bool DEFAULT: False

arguments

Additional arguments for the exchange. Defaults to None.

TYPE: Optional[AnyDict] DEFAULT: None

timeout

Timeout for the operation. Defaults to None.

TYPE: TimeoutType DEFAULT: None

robust

Whether to use robust mode for the exchange. Defaults to True.

TYPE: bool DEFAULT: True

bind_to

Exchange to bind to. Defaults to None.

TYPE: Optional[RabbitExchange] DEFAULT: None

bind_arguments

Arguments for the binding. Defaults to None.

TYPE: Optional[AnyDict] DEFAULT: None

routing_key

Routing key for the exchange. Defaults to "".

TYPE: str DEFAULT: ''

RAISES DESCRIPTION
NotImplementedError

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/rabbit/shared/schemas.py
def __init__(
    self,
    name: str,
    type: ExchangeType = ExchangeType.DIRECT,
    durable: bool = False,
    auto_delete: bool = False,
    internal: bool = False,
    passive: bool = False,
    arguments: Optional[AnyDict] = None,
    timeout: TimeoutType = None,
    robust: bool = True,
    bind_to: Optional["RabbitExchange"] = None,
    bind_arguments: Optional[AnyDict] = None,
    routing_key: str = "",
):
    """Initialize a RabbitExchange object.

    Args:
        name (str): Name of the exchange.
        type (ExchangeType, optional): Type of the exchange. Defaults to ExchangeType.DIRECT.
        durable (bool, optional): Whether the exchange should survive broker restarts. Defaults to False.
        auto_delete (bool, optional): Whether the exchange should be deleted when no longer in use. Defaults to False.
        internal (bool, optional): Whether the exchange is used for internal purposes and should not be published to directly. Defaults to False.
        passive (bool, optional): Whether to check if the exchange exists before creating it. Defaults to False.
        arguments (Optional[AnyDict], optional): Additional arguments for the exchange. Defaults to None.
        timeout (TimeoutType, optional): Timeout for the operation. Defaults to None.
        robust (bool, optional): Whether to use robust mode for the exchange. Defaults to True.
        bind_to (Optional["RabbitExchange"], optional): Exchange to bind to. Defaults to None.
        bind_arguments (Optional[AnyDict], optional): Arguments for the binding. Defaults to None.
        routing_key (str, optional): Routing key for the exchange. Defaults to "".

    Raises:
        NotImplementedError:
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    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=name,
        type=type.value,
        durable=durable,
        auto_delete=auto_delete,
        routing_key=routing_key,
        bind_to=bind_to,
        bind_arguments=bind_arguments,
        robust=robust,
        internal=internal,
        passive=passive,
        timeout=timeout,
        arguments=arguments,
    )

arguments class-attribute instance-attribute #

arguments: Optional[AnyDict] = None

auto_delete class-attribute instance-attribute #

auto_delete: bool = False

bind_arguments class-attribute instance-attribute #

bind_arguments: Optional[AnyDict] = None

bind_to class-attribute instance-attribute #

bind_to: Optional[RabbitExchange] = None

durable class-attribute instance-attribute #

durable: bool = False

internal class-attribute instance-attribute #

internal: bool = False

name class-attribute instance-attribute #

name: str = Field(...)

passive class-attribute instance-attribute #

passive: bool = False

robust class-attribute instance-attribute #

robust: bool = True

routing_key class-attribute instance-attribute #

routing_key: str = ''

timeout class-attribute instance-attribute #

timeout: TimeoutType = None

type class-attribute instance-attribute #

type: str = ExchangeType.DIRECT.value

validate classmethod #

validate(
    value: Union[str, NameRequiredCls, None]
) -> Optional[NameRequiredCls]

Validates a value.

PARAMETER DESCRIPTION
value

The value to be validated.

TYPE: Union[str, NameRequiredCls, None]

RETURNS DESCRIPTION
Optional[NameRequiredCls]

The validated value.

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/broker/schemas.py
@classmethod
def validate(
    cls: Type[NameRequiredCls], value: Union[str, NameRequiredCls, None]
) -> Optional[NameRequiredCls]:
    """Validates a value.

    Args:
        value: The value to be validated.

    Returns:
        The validated value.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    if value is not None:
        if isinstance(value, str):
            value = cls(value)
    return value

Last update: 2023-11-13