Skip to content

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