Skip to content

get_watcher

faststream.broker.utils.get_watcher #

get_watcher(
    logger: Optional[logging.Logger],
    try_number: Union[bool, int] = True,
) -> BaseWatcher

Get a watcher object based on the provided parameters.

PARAMETER DESCRIPTION
logger

Optional logger object for logging messages.

TYPE: Optional[Logger]

try_number

Optional parameter to specify the type of watcher. - If set to True, an EndlessWatcher object will be returned. - If set to False, a OneTryWatcher object will be returned. - If set to an integer, a CounterWatcher object with the specified maximum number of tries will be returned.

TYPE: Union[bool, int] DEFAULT: True

RETURNS DESCRIPTION
BaseWatcher

A watcher object based on the provided parameters.

Note

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

Source code in faststream/broker/utils.py
def get_watcher(
    logger: Optional[logging.Logger],
    try_number: Union[bool, int] = True,
) -> BaseWatcher:
    """Get a watcher object based on the provided parameters.

    Args:
        logger: Optional logger object for logging messages.
        try_number: Optional parameter to specify the type of watcher.
            - If set to True, an EndlessWatcher object will be returned.
            - If set to False, a OneTryWatcher object will be returned.
            - If set to an integer, a CounterWatcher object with the specified maximum number of tries will be returned.

    Returns:
        A watcher object based on the provided parameters.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    watcher: Optional[BaseWatcher]
    if try_number is True:
        watcher = EndlessWatcher()
    elif try_number is False:
        watcher = OneTryWatcher()
    else:
        watcher = CounterWatcher(logger=logger, max_tries=try_number)
    return watcher

Last update: 2023-11-13