Skip to content

OneTryWatcher

faststream.broker.push_back_watcher.OneTryWatcher #

OneTryWatcher(
    max_tries: int = 0, logger: Optional[Logger] = None
)

Bases: BaseWatcher

Initialize the class.

PARAMETER DESCRIPTION
max_tries

Maximum number of tries allowed

TYPE: int DEFAULT: 0

logger

Optional logger object

TYPE: Optional[Logger] DEFAULT: None

RAISES DESCRIPTION
NotImplementedError

If the method is not implemented in the subclass.
Source code in faststream/broker/push_back_watcher.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    max_tries: int = 0,
    logger: Optional[Logger] = None,
) -> None:
    """Initialize the class.

    Args:
        max_tries: Maximum number of tries allowed
        logger: Optional logger object

    Raises:
        NotImplementedError: If the method is not implemented in the subclass.

    """
    self.logger = logger
    self.max_tries = max_tries

logger instance-attribute #

logger = logger

max_tries instance-attribute #

max_tries: int = max_tries

add #

add(message_id: str) -> None

Add a message.

PARAMETER DESCRIPTION
message_id

ID of the message to be added

TYPE: str

RETURNS DESCRIPTION
None

None
Source code in faststream/broker/push_back_watcher.py 
145
146
147
148
149
150
151
152
153
154
155
def add(self, message_id: str) -> None:
    """Add a message.

    Args:
        message_id: ID of the message to be added

    Returns:
        None

    """
    pass

is_max #

is_max(message_id: str) -> bool

Check if the given message ID is the maximum.

PARAMETER DESCRIPTION
message_id

The ID of the message to check.

TYPE: str

RETURNS DESCRIPTION
bool

True if the given message ID is the maximum, False otherwise.
Source code in faststream/broker/push_back_watcher.py 
157
158
159
160
161
162
163
164
165
166
167
def is_max(self, message_id: str) -> bool:
    """Check if the given message ID is the maximum.

    Args:
        message_id: The ID of the message to check.

    Returns:
        True if the given message ID is the maximum, False otherwise.

    """
    return True

remove #

remove(message_id: str) -> None

Remove a message.

PARAMETER DESCRIPTION
message_id

ID of the message to be removed

TYPE: str

RETURNS DESCRIPTION
None

None
Source code in faststream/broker/push_back_watcher.py 
169
170
171
172
173
174
175
176
177
178
179
def remove(self, message_id: str) -> None:
    """Remove a message.

    Args:
        message_id: ID of the message to be removed

    Returns:
        None

    """
    pass