Listeners reference#

WhatsApp.listen(to, *, filters=None, cancelers=None, timeout=None)

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 (BaseListenerIdentifier) – The identifier of the update to listen to.

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

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

  • timeout (float | None) – The time to wait for the update, raise ListenerTimeout if the time passes

Returns:

The update that passed the filters

Raises:
Return type:

BaseUpdate

WhatsApp.stop_listening(to, *, reason=None)

Stop listening to updates for a specific listener

  • Raising ListenerStopped to the listener

Parameters:
  • to (BaseListenerIdentifier) – The identifier of the listener to stop

  • reason (str | None) – The reason to stop listening

Raises:

ValueError – If the listener does not exist

SentMessage.wait_for_reply(*, force_quote=False, filters=None, cancelers=None, ignore_updates=True, timeout=None)#

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 (bool) – Whether to force the reply to quote the sent message.

  • filters (Filter) – The filters to apply to the reply.

  • cancelers (Filter) – The filters to cancel the listening.

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

  • timeout (float | None) – The time to wait for a reply.

Returns:

The reply message.

Raises:
Return type:

Message

SentMessage.wait_for_click(*, filters=None, cancelers=None, ignore_updates=True, timeout=None)#

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 (Filter) – The filters to apply to the button click.

  • cancelers (Filter) – The filters to cancel the listening.

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

  • timeout (float | None) – The time to wait for the button click.

Returns:

The clicked button.

Raises:
Return type:

CallbackButton

SentMessage.wait_for_selection(*, filters=None, cancelers=None, ignore_updates=True, timeout=None)#

Wait for a callback selection.

Parameters:
  • filters (Filter) – The filters to apply to the selection.

  • cancelers (Filter) – The filters to cancel the listening.

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

  • timeout (float | None) – The time to wait for the selection.

Returns:

The callback selection.

Raises:
Return type:

CallbackSelection

SentMessage.wait_until_read(*, cancel_on_new_update=False, cancel_if_failed=True, cancelers=None, timeout=None)#

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 (bool) – Whether to cancel when another message/button click arrives (which may indicate the previous message was read).

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

  • cancelers (Filter) – The filters to cancel the listening.

  • timeout (float | None) – The time to wait for the message to be read.

Returns:

The message status.

Raises:
Return type:

MessageStatus

SentMessage.wait_until_delivered(*, cancel_if_failed=True, cancelers=None, timeout=None)#

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 (bool) – Whether to cancel the listener if the message failed to send.

  • cancelers (Filter) – The filters to cancel the listening.

  • timeout (float | None) – The time to wait for the message to be delivered.

Returns:

The message status.

Raises:
Return type:

MessageStatus

SentMessage.wait_for_completion(*, filters=None, cancelers=None, ignore_updates=True, timeout=None)#

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 (Filter) – The filters to apply to the completion.

  • cancelers (Filter) – The filters to cancel the listening.

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

  • timeout (float | None) – The time to wait for the completion.

Returns:

The flow completion.

Raises:
Return type:

FlowCompletion

SentMessage.wait_for_call_permission(*, filters=None, cancelers=None, ignore_updates=True, timeout=None)#

Wait for a call permission update.

Example

@wa.on_message(filters.command("start"))
def start(w: WhatsApp, m: Message):
    r = m.reply(
        text="Do you allow calls?",
        buttons=types.CallPermissionRequestButton(),
    )
    call_permission = r.wait_for_call_permission()
    if call_permission:
        r.reply("You allowed calls", quote=True)
    else:
        r.reply("You denied calls", quote=True)
Parameters:
  • filters (Filter) – The filters to apply to the call permission update.

  • cancelers (Filter) – The filters to cancel the listening.

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

  • timeout (float | None) – The time to wait for the call permission update.

Returns:

The call permission update.

Raises:
Return type:

CallPermissionUpdate

SentMessage.wait_for_incoming_voice_call(*, filters=None, cancelers=None, ignore_updates=True, timeout=None)#

Wait for an incoming call in response to a voice call button.

Example

@wa.on_message(filters.command("start"))
def start(w: WhatsApp, m: Message):
    r = m.reply(
        text="Click the button to call me",
        buttons=types.VoiceCallButton(),
    )
    call_connect = r.wait_for_incoming_voice_call()
    r.reply("You called me!", quote=True)
Parameters:
  • filters (Filter) – The filters to apply to the incoming call.

  • cancelers (Filter) – The filters to cancel the listening.

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

  • timeout (float | None) – The time to wait for the incoming call.

Returns:

The call connect update.

Raises:
Return type:

CallConnect

SentVoiceMessage.wait_until_played(*, filters=None, cancelers=None, timeout=None)#

Wait for the voice message to be played by the recipient.

Example

@wa.on_message(filters.command("start"))
def start(w: WhatsApp, m: Message):
    r = m.reply_voice(
        voice_url="https://example.com/voice.mp3",
    )
    r.wait_until_played()
    r.reply("You played the voice message", quote=True)
Parameters:
  • filters (Filter) – The filters to apply to the played status.

  • cancelers (Filter) – The filters to cancel the listening.

  • timeout (float | None) – The time to wait for the voice message to be played.

Returns:

The message status.

Raises:
Return type:

MessageStatus

SentLocationRequest.wait_for_location(*, force_current_location=True, filters=None, cancelers=None, ignore_updates=True, timeout=None)#

Wait for a location message in response to the location request.

Parameters:
  • force_current_location (bool) – Whether to only accept current location messages.

  • filters (Filter) – The filters to apply to the location message.

  • cancelers (Filter) – The filters to cancel the listening.

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

  • timeout (float | None) – The time to wait for the location message.

Returns:

The location message.

Raises:
Return type:

Message

CreatedTemplate.wait_until_approved(*, cancel_on_rejection=True, cancelers=None, timeout=None)#

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 (bool) – Whether to cancel the waiting process if the template is rejected. Defaults to True.

  • cancelers (pywa_filters.Filter) – A filter to cancel the waiting process.

  • timeout (float | None) – 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=”…”))