get_watcher

faststream.broker.acknowledgement_watcher.get_watcher

get_watcher(logger, try_number)

Get a watcher object based on the provided parameters.

PARAMETER	DESCRIPTION
logger	Optional logger object for logging messages. TYPE: Optional[LoggerProto]
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]

Source code in faststream/broker/acknowledgement_watcher.py

def get_watcher(
    logger: Optional["LoggerProto"],
    try_number: Union[bool, int],
) -> 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.
    """
    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