Skip to content

get_watcher

faststream.broker.utils.get_watcher #

get_watcher(
    logger: Optional[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.
Source code in faststream/broker/utils.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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.

    """
    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