Listeners reference

Listeners reference#

WhatsApp.listen(to: BaseListenerIdentifier, *, filters: Filter = None, cancelers: Filter = None, timeout: float | None = None) → BaseUpdate

Listen to an update.

Example

try:
    wa.send_message(
        to="123456",
        text="Send me a message",
        buttons=[Button(title="Cancel", callback_data="cancel")]
    )
    update: Message = wa.listen(
        to=UserUpdateListenerIdentifier(sender="123456", recipient="654321"),
        filters=filters.message & filters.text,
        cancelers=filters.callback_button & filters.matches("cancel"),
        timeout=10
    )
    print(update)
except ListenerTimeout:
    print("Listener timed out")
except ListenerCanceled:
    print("Listener was canceled")
except ListenerStopped:
    print("Listener was stopped")
Parameters:

  • to – The identifier of the update to listen to.

  • filters – The filters to apply to the update, return the update if the filters pass.

  • cancelers – The filters to cancel the listening, raise ListenerCanceled if the update matches.

  • timeout – The time to wait for the update, raise ListenerTimeout if the time passes

Returns:

The update that passed the filters

Raises:
WhatsApp.stop_listening(to: BaseListenerIdentifier, *, reason: str | None = None) → None

Stop listening to updates for a specific listener

  • Raising ListenerStopped to the listener

Parameters:

  • to – The identifier of the listener to stop

  • reason – The reason to stop listening

Raises:

ValueError – If the listener does not exist

SentMessage.wait_for_reply(*, force_quote: bool = False, filters: Filter = None, cancelers: Filter = None, ignore_updates: bool = True, timeout: float | None = None) → Message#

Wait for a message reply to the sent message.

  • Shortcut for listen() with filters=filters.message.

Example

@wa.on_message(filters.command("start"))
def start(w: WhatsApp, m: Message):
    user_id: str = m.reply(
        text=f"Hi {m.from_user.name}! Please enter your ID",
        buttons=[Button(title="Cancel", callback_data="cancel")],
    ).wait_for_reply(
        filters=filters.text & filters.new(lambda _, m: m.text.isdigit()),
        cancelers=filters.callback_button & filters.matches("cancel"),
    ).text
    ...
Parameters:

  • force_quote – Whether to force the reply to quote the sent message.

  • filters – The filters to apply to the reply.

  • cancelers – The filters to cancel the listening.

  • ignore_updates – Whether to ignore user updates that do not pass the filters.

  • timeout – The time to wait for a reply.

Returns:

The reply message.

Raises:
SentMessage.wait_for_click(*, filters: Filter = None, cancelers: Filter = None, ignore_updates: bool = True, timeout: float | None = None) → CallbackButton#

Wait for a button click.

Example

@wa.on_message(filters.command("start"))
def start(w: WhatsApp, m: Message):
    r = m.reply(
        text="Click a button",
        buttons=[
            Button(title="Option 1", callback_data="option1"),
            Button(title="Option 2", callback_data="option2"),
        ],
    )
    option = r.wait_for_click()
    r.reply(f"You clicked: {option.title}", quote=True)
Parameters:

  • filters – The filters to apply to the button click.

  • cancelers – The filters to cancel the listening.

  • ignore_updates – Whether to ignore user updates that do not pass the filters.

  • timeout – The time to wait for the button click.

Returns:

The clicked button.

Raises:
SentMessage.wait_for_selection(*, filters: Filter = None, cancelers: Filter = None, ignore_updates: bool = True, timeout: float | None = None) → CallbackSelection#

Wait for a callback selection.

Parameters:

  • filters – The filters to apply to the selection.

  • cancelers – The filters to cancel the listening.

  • ignore_updates – Whether to ignore user updates that do not pass the filters.

  • timeout – The time to wait for the selection.

Returns:

The callback selection.

Raises:
SentMessage.wait_for_completion(*, filters: Filter = None, cancelers: Filter = None, ignore_updates: bool = True, timeout: float | None = None) → FlowCompletion#

Wait for a flow completion.

Example

@wa.on_message(filters.command("start"))
def start(w: WhatsApp, m: Message):
    flow_completion = m.reply(
        text="Answer the questions",
        buttons=FlowButton(flow_token=..., ...),
    ).wait_for_completion()
Parameters:

  • filters – The filters to apply to the completion.

  • cancelers – The filters to cancel the listening.

  • ignore_updates – Whether to ignore user updates that do not pass the filters.

  • timeout – The time to wait for the completion.

Returns:

The flow completion.

Raises:
SentMessage.wait_until_read(*, cancel_on_new_update: bool = False, cancel_if_failed: bool = True, cancelers: Filter = None, timeout: float | None = None) → MessageStatus#

Wait for the message to be read by the recipient.

  • Shortcut for listen() with filters=filters.message_status & filters.read.

Note: This method will not work if the recipient has disabled read receipts. make sure to use cancel_on_new_update=True to cancel the listening if the message is probably read, or use a timeout / cancelers.

Example

@wa.on_message(filters.command("start"))
def start(w: WhatsApp, m: Message):
    r = m.reply("This message waits for you to read it")
    try:
        r.wait_until_read(cancel_on_new_update=True)
    except ListenerCanceled as e:
        print(e.update) # The update that canceled the listener
        r.reply("You turned off read receipts")
    r.reply("You read this message", quote=True)
Parameters:

  • cancel_on_new_update – Whether to cancel when another message/button click arrives (which may indicate the previous message was read).

  • cancel_if_failed – Whether to cancel the listener if the message failed to send.

  • cancelers – The filters to cancel the listening.

  • timeout – The time to wait for the message to be read.

Returns:

The message status.

Raises:
SentMessage.wait_until_delivered(*, cancel_if_failed: bool = True, cancelers: Filter = None, timeout: float | None = None) → MessageStatus#

Wait for the message to be delivered to the recipient.

Example

@wa.on_message(filters.command("start"))
def start(w: WhatsApp, m: Message):
    r = m.reply("This message waits for you to receive it")
    r.wait_until_delivered()
    r.reply("You received the message", quote=True)
Parameters:

  • cancel_if_failed – Whether to cancel the listener if the message failed to send.

  • cancelers – The filters to cancel the listening.

  • timeout – The time to wait for the message to be delivered.

Returns:

The message status.

Raises:
CreatedTemplate.wait_until_approved(*, cancel_on_rejection: bool = True, cancelers: pywa_filters.Filter = None, timeout: float | None = None) → TemplateStatusUpdate#

Wait until the template is approved.

Example usage:

>>> from pywa import WhatsApp, filters
>>> wa = WhatsApp(...)
>>> created_template = wa.create_template(...)
>>> status = created_template.wait_until_approved(cancelers=filters.template_status & filters.template_status_rejected)
>>> print(f"Template {created_template.id} is approved with status: {status.new_status}")
Parameters:

  • cancel_on_rejection – Whether to cancel the waiting process if the template is rejected. Defaults to True.

  • cancelers – A filter to cancel the waiting process.

  • timeout – The maximum time to wait for the template to be approved.

Returns:

An update containing the status of the template once it is approved.

Return type:

TemplateStatusUpdate

class pywa.listeners.ListenerTimeout#

The listener timed out

Example

try:
    wa.listen(..., timeout=10)
except ListenerTimeout as e:
    wa.send_message(to="123456", text="You took too long to respond")
Variables:

timeout – The timeout that was set for the listener

class pywa.listeners.ListenerCanceled#

The listener was canceled by a filter

Example

try:
    wa.listen(
        to=UserUpdateListenerIdentifier(
            sender="123456",
            recipient="654321"
        ),
        filters=filters.message & filters.text,
        cancelers=filters.callback_button & filters.matches("cancel")
    )
except ListenerCanceled as e:
    assert e.update.data == "cancel" # the update that caused the listener to be canceled
    wa.send_message("123456", "You cancelled the listener by clicking the `cancel` button")
Variables:

update (BaseUpdate | BaseUserUpdate | None) – The update that caused the listener to be canceled

class pywa.listeners.ListenerStopped#

The listener was stopped manually by wa.stop_listening(…)

Example

try:
    wa.listen(...)
except ListenerStopped as e:
    print(e.reason) # print the reason the listener was stopped
    wa.send_message("123456", "The listener was stopped")
Variables:

reason – The reason the listener was stopped (set by wa.stop_listening(reason=”…”))

Contents