From 893efa9f22deba6fcb952e4662842a7bf3124d7f Mon Sep 17 00:00:00 2001 From: David Lev Date: Wed, 24 Apr 2024 18:10:00 +0300 Subject: [PATCH] [client,message_status] allowing to pass `CallbackData` subclasses to`tracker` --- pywa/api.py | 328 ++--------------------- pywa/client.py | 486 +++++++++++++++++++++-------------- pywa/handlers.py | 88 +++++-- pywa/types/base_update.py | 66 ++--- pywa/types/callback.py | 18 +- pywa/types/message.py | 6 +- pywa/types/message_status.py | 71 ++++- pywa/types/others.py | 18 ++ pywa/types/template.py | 6 +- 9 files changed, 504 insertions(+), 583 deletions(-) diff --git a/pywa/api.py b/pywa/api.py index 5e637a3..a51ae8b 100644 --- a/pywa/api.py +++ b/pywa/api.py @@ -162,53 +162,6 @@ def set_business_public_key( data={"business_public_key": public_key}, ) - def send_text_message( - self, - to: str, - text: str, - preview_url: bool = False, - reply_to_message_id: str | None = None, - tracker: str | None = None, - ) -> dict[str, dict | list]: - """ - Send a text message to a WhatsApp user. - - Return example:: - - { - 'messaging_product': 'whatsapp', - 'contacts': [{'input': '1234567890', 'wa_id': '1234567890'}], - 'messages': [{'id': 'wamid.XXXXXXXXXXXXXXXX=='}] - } - - Args: - to: The WhatsApp ID of the recipient. - text: The text to send. - preview_url: Whether to show a preview of the URL in the message. - reply_to_message_id: The ID of the message to reply to. - tracker: The data that you can track by MessageStatus. - - Returns: - The sent message. - """ - data = { - **self._common_keys, - "to": to, - "type": "text", - "text": {"body": text, "preview_url": preview_url}, - } - if reply_to_message_id: - data["context"] = {"message_id": reply_to_message_id} - - if tracker: - data["biz_opaque_callback_data"] = tracker - - return self._make_request( - method="POST", - endpoint=f"/{self.phone_id}/messages", - json=data, - ) - def upload_media( self, media: bytes, @@ -309,142 +262,6 @@ def delete_media(self, media_id: str) -> dict[str, bool]: """ return self._make_request(method="DELETE", endpoint=f"/{media_id}") - def send_media( - self, - to: str, - media_id_or_url: str, - media_type: str, - is_url: bool, - caption: str | None = None, - filename: str | None = None, - tracker: str | None = None, - ) -> dict[str, dict | list]: - """ - Send a media file to a WhatsApp user. - - Return example:: - - { - 'messaging_product': 'whatsapp', - 'contacts': [{'input': '1234567890', 'wa_id': '1234567890'}], - 'messages': [{'id': 'wamid.XXXXXXXXXXXXXXXX=='}] - } - - Args: - to: The WhatsApp ID of the recipient. - media_id_or_url: The ID or URL of the media file to send. - is_url: Whether the media_id_or_url is a URL or an ID. - media_type: The type of the media file (e.g. 'image', 'video', 'document'). - caption: The caption to send with the media file (only for images, videos and documents). - filename: The filename to send with the media file (only for documents). - tracker: The data that you can track by MessageStatus. - - Returns: - The sent message. - """ - data = { - **self._common_keys, - "to": to, - "type": media_type, - media_type: { - ("link" if is_url else "id"): media_id_or_url, - **({"caption": caption} if caption else {}), - **({"filename": filename} if filename else {}), - }, - **({"biz_opaque_callback_data": tracker} if tracker else {}), - } - return self._make_request( - method="POST", - endpoint=f"/{self.phone_id}/messages", - json=data, - ) - - def send_reaction( - self, - to: str, - emoji: str, - message_id: str, - ) -> dict[str, dict | list]: - """ - Send a reaction to a message. - - Return example:: - - { - 'messaging_product': 'whatsapp', - 'contacts': [{'input': '1234567890', 'wa_id': '1234567890'}], - 'messages': [{'id': 'wamid.XXXXXXXXXXXXXXXX=='}] - } - - Args: - to: The WhatsApp ID of the recipient. - emoji: The emoji to react with (empty to remove reaction). - message_id: The ID of the message to react to. - - Returns: - The sent message. - """ - return self._make_request( - method="POST", - endpoint=f"/{self.phone_id}/messages/", - json={ - **self._common_keys, - "to": to, - "type": "reaction", - "reaction": {"emoji": emoji, "message_id": message_id}, - }, - ) - - def send_location( - self, - to: str, - latitude: float, - longitude: float, - name: str | None = None, - address: str | None = None, - tracker: str | None = None, - ) -> dict[str, dict | list]: - """ - Send a location to a WhatsApp user. - - Return example:: - - { - 'messaging_product': 'whatsapp', - 'contacts': [{'input': '1234567890', 'wa_id': '1234567890'}], - 'messages': [{'id': 'wamid.XXXXXXXXXXXXXXXX=='}] - } - - Args: - to: The WhatsApp ID of the recipient. - latitude: The latitude of the location. - longitude: The longitude of the location. - name: The name of the location. - address: The address of the location. - tracker: The data that you can track by MessageStatus. - - Returns: - The sent message. - """ - data = { - **self._common_keys, - "to": to, - "type": "location", - "location": { - "latitude": latitude, - "longitude": longitude, - "name": name, - "address": address, - }, - **({"biz_opaque_callback_data": tracker} if tracker else {}), - } - - return self._make_request( - method="POST", - endpoint=f"/{self.phone_id}/messages", - json=data, - ) - def send_raw_request(self, method: str, endpoint: str, **kwargs) -> Any: """ Send a raw request to WhatsApp Cloud API. @@ -481,62 +298,41 @@ def send_raw_request(self, method: str, endpoint: str, **kwargs) -> Any: **kwargs, ) - def send_interactive_message( + def send_message( self, to: str, - type_: str, - action: dict[str, Any], - header: dict | None = None, - body: str | None = None, - footer: str | None = None, + typ: str, + msg: dict[str, str | list[str]] | tuple[dict], reply_to_message_id: str | None = None, tracker: str | None = None, ) -> dict[str, dict | list]: """ - Send an interactive message to a WhatsApp user. - - Return example:: - - { - 'messaging_product': 'whatsapp', - 'contacts': [{'input': '1234567890', 'wa_id': '1234567890'}], - 'messages': [{'id': 'wamid.XXXXXXXXXXXXXXXX=='}] - } + Send a message to a WhatsApp user. Args: - to: The WhatsApp ID of the recipient. - type_: The type of the message (e.g. ``list``, ``button``, ``product``, etc.). - action: The action of the message. - header: The header of the message. - body: The body of the message. - footer: The footer of the message. + to: The phone number to send the message to. + typ: The type of the message (e.g. ``text``, ``image``, etc.). + msg: The message object to send. reply_to_message_id: The ID of the message to reply to. - tracker: The data that you can track by MessageStatus. + tracker: The tracker to send with the message. Returns: - The sent message. + The response from the WhatsApp Cloud API. """ + data = { + **self._common_keys, + "to": to, + "type": typ, + typ: msg, + } + if reply_to_message_id: + data["context"] = {"message_id": reply_to_message_id} + if tracker: + data["biz_opaque_callback_data"] = tracker return self._make_request( method="POST", endpoint=f"/{self.phone_id}/messages", - json={ - **self._common_keys, - "to": to, - "type": "interactive", - "interactive": { - "type": type_, - "action": action, - **({"header": header} if header else {}), - **({"body": {"text": body}} if body else {}), - **({"footer": {"text": footer}} if footer else {}), - }, - **( - {"context": {"message_id": reply_to_message_id}} - if reply_to_message_id - else {} - ), - **({"biz_opaque_callback_data": tracker} if tracker else {}), - }, + json=data, ) def register_phone_number( @@ -566,49 +362,6 @@ def register_phone_number( }, ) - def send_contacts( - self, - to: str, - contacts: tuple[dict[str, Any], ...], - reply_to_message_id: str | None = None, - tracker: str | None = None, - ) -> dict[str, dict | list]: - """ - Send a list of contacts to a WhatsApp user. - - Return example:: - - { - 'messaging_product': 'whatsapp', - 'contacts': [{'input': '1234567890', 'wa_id': '1234567890'}], - 'messages': [{'id': 'wamid.XXXXXXXXXXXXXXXX=='}] - } - - Args: - to: The WhatsApp ID of the recipient. - contacts: The contacts to send. - reply_to_message_id: The ID of the message to reply to. - tracker: The data that you can track by MessageStatus. - - Returns: - The sent message. - """ - data = { - **self._common_keys, - "to": to, - "type": "contacts", - "contacts": tuple(contacts), - } - if reply_to_message_id: - data["context"] = {"message_id": reply_to_message_id} - if tracker: - data["biz_opaque_callback_data"] = tracker - return self._make_request( - method="POST", - endpoint=f"/{self.phone_id}/messages", - json=data, - ) - def mark_message_as_read(self, message_id: str) -> dict[str, bool]: """ Mark a message as read. @@ -761,47 +514,6 @@ def create_template( json=template, ) - def send_template( - self, - to: str, - template: dict, - reply_to_message_id: str | None = None, - tracker: str | None = None, - ) -> dict[str, dict | list]: - """ - Send a template to a WhatsApp user. - - Args: - to: The WhatsApp ID of the recipient. - template: The template to send. - reply_to_message_id: The ID of the message to reply to. - tracker: The data that you can track by MessageStatus. - - Returns example:: - - { - 'messaging_product': 'whatsapp', - 'contacts': [{'input': '1234567890', 'wa_id': '1234567890'}], - 'messages': [{'id': 'wamid.XXXXXXXXXXXXXXXX=='}] - } - - """ - data = { - **self._common_keys, - "to": to, - "type": "template", - "template": template, - } - if reply_to_message_id: - data["context"] = {"message_id": reply_to_message_id} - if tracker: - data["biz_opaque_callback_data"] = tracker - return self._make_request( - method="POST", - endpoint=f"/{self.phone_id}/messages", - json=data, - ) - def create_flow( self, business_account_id: str, diff --git a/pywa/client.py b/pywa/client.py index e810d73..9df3d46 100644 --- a/pywa/client.py +++ b/pywa/client.py @@ -11,7 +11,7 @@ import os import pathlib import warnings -from typing import BinaryIO, Iterable, Literal +from typing import BinaryIO, Iterable, Literal, Any import requests @@ -33,8 +33,11 @@ Template, TemplateResponse, FlowButton, + ChatOpened, + MessageType, ) from pywa.types.base_update import BaseUpdate +from pywa.types.callback import CallbackDataT, CallbackData from pywa.types.flows import ( FlowCategory, FlowJSON, @@ -42,6 +45,7 @@ FlowValidationError, FlowAsset, ) +from pywa.types.others import InteractiveType from pywa.utils import FastAPI, Flask from pywa.server import Server @@ -247,7 +251,7 @@ def send_message( preview_url: bool = False, reply_to_message_id: str | None = None, keyboard: None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send a message to a WhatsApp user. @@ -358,7 +362,7 @@ def send_message( preview_url: Whether to show a preview of the URL in the message (if any). reply_to_message_id: The message ID to reply to (optional). keyboard: Deprecated and will be removed in a future version, use ``buttons`` instead. - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent message. @@ -374,27 +378,31 @@ def send_message( ) if not buttons: - return self.api.send_text_message( + return self.api.send_message( to=str(to), - text=text, - preview_url=preview_url, + typ=MessageType.TEXT.value, + msg={"body": text, "preview_url": preview_url}, reply_to_message_id=reply_to_message_id, - tracker=tracker, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] - type_, kb = _resolve_buttons_param(buttons) - return self.api.send_interactive_message( + typ, kb = _resolve_buttons_param(buttons) + return self.api.send_message( to=str(to), - type_=type_, - action=kb, - header={ - "type": "text", - "text": header, - } - if header - else None, - body=text, - footer=footer, - tracker=tracker, + typ=MessageType.INTERACTIVE.value, + msg=_get_interactive_msg( + typ=typ, + action=kb, + header={ + "type": MessageType.TEXT.value, + "text": header, + } + if header + else None, + body=text, + footer=footer, + ), + reply_to_message_id=reply_to_message_id, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] send_text = send_message # alias @@ -409,7 +417,7 @@ def send_image( buttons: Iterable[Button] | ButtonUrl | FlowButton | None = None, reply_to_message_id: str | None = None, mime_type: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send an image to a WhatsApp user. @@ -437,7 +445,7 @@ def send_image( mime_type: The mime type of the image (optional, required when sending an image as bytes or a file object, or file path that does not have an extension). body: Deprecated and will be removed in a future version, use ``caption`` instead. - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent image message. @@ -460,33 +468,38 @@ def send_image( filename=None, ) if not buttons: - return self.api.send_media( + return self.api.send_message( to=str(to), - media_id_or_url=image, - is_url=is_url, - media_type="image", - caption=caption, - tracker=tracker, + typ=MessageType.IMAGE.value, + msg=_get_media_msg( + media_id_or_url=image, + is_url=is_url, + caption=caption, + ), + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] if not caption: raise ValueError( "A caption must be provided when sending an image with buttons." ) - type_, kb = _resolve_buttons_param(buttons) - return self.api.send_interactive_message( + typ, kb = _resolve_buttons_param(buttons) + return self.api.send_message( to=str(to), - type_=type_, - action=kb, - header={ - "type": "image", - "image": { - "link" if is_url else "id": image, + typ=MessageType.INTERACTIVE.value, + msg=_get_interactive_msg( + typ=typ, + action=kb, + header={ + "type": MessageType.IMAGE.value, + "image": { + "link" if is_url else "id": image, + }, }, - }, - body=caption, - footer=footer, + body=caption, + footer=footer, + ), reply_to_message_id=reply_to_message_id, - tracker=tracker, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def send_video( @@ -499,7 +512,7 @@ def send_video( buttons: Iterable[Button] | ButtonUrl | FlowButton | None = None, reply_to_message_id: str | None = None, mime_type: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send a video to a WhatsApp user. @@ -528,7 +541,7 @@ def send_video( mime_type: The mime type of the video (optional, required when sending a video as bytes or a file object, or file path that does not have an extension). body: Deprecated and will be removed in a future version, use ``caption`` instead. - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent video. @@ -551,33 +564,38 @@ def send_video( filename=None, ) if not buttons: - return self.api.send_media( + return self.api.send_message( to=str(to), - media_id_or_url=video, - is_url=is_url, - media_type="video", - caption=caption, - tracker=tracker, + typ=MessageType.VIDEO.value, + msg=_get_media_msg( + media_id_or_url=video, + is_url=is_url, + caption=caption, + ), + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] if not caption: raise ValueError( "A caption must be provided when sending a video with buttons." ) - type_, kb = _resolve_buttons_param(buttons) - return self.api.send_interactive_message( + typ, kb = _resolve_buttons_param(buttons) + return self.api.send_message( to=str(to), - type_=type_, - action=kb, - header={ - "type": "video", - "video": { - "link" if is_url else "id": video, + typ=MessageType.INTERACTIVE.value, + msg=_get_interactive_msg( + typ=typ, + action=kb, + header={ + "type": MessageType.VIDEO.value, + "video": { + "link" if is_url else "id": video, + }, }, - }, - body=caption, - footer=footer, + body=caption, + footer=footer, + ), reply_to_message_id=reply_to_message_id, - tracker=tracker, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def send_document( @@ -591,7 +609,7 @@ def send_document( buttons: Iterable[Button] | ButtonUrl | FlowButton | None = None, reply_to_message_id: str | None = None, mime_type: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send a document to a WhatsApp user. @@ -622,7 +640,7 @@ def send_document( mime_type: The mime type of the document (optional, required when sending a document as bytes or a file object, or file path that does not have an extension). body: Deprecated and will be removed in a future version, use ``caption`` instead. - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent document. @@ -645,35 +663,39 @@ def send_document( media_type=None, ) if not buttons: - return self.api.send_media( + return self.api.send_message( to=str(to), - media_id_or_url=document, - is_url=is_url, - media_type="document", - caption=caption, - filename=filename, - tracker=tracker, + typ=MessageType.DOCUMENT.value, + msg=_get_media_msg( + media_id_or_url=document, + is_url=is_url, + caption=caption, + ), + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] if not caption: raise ValueError( "A caption must be provided when sending a document with buttons." ) type_, kb = _resolve_buttons_param(buttons) - return self.api.send_interactive_message( + return self.api.send_message( to=str(to), - type_=type_, - action=kb, - header={ - "type": "document", - "document": { - "link" if is_url else "id": document, - "filename": filename, + typ=MessageType.INTERACTIVE.value, + msg=_get_interactive_msg( + typ=type_, + action=kb, + header={ + "type": MessageType.DOCUMENT.value, + "document": { + "link" if is_url else "id": document, + "filename": filename, + }, }, - }, - body=caption, - footer=footer, + body=caption, + footer=footer, + ), reply_to_message_id=reply_to_message_id, - tracker=tracker, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def send_audio( @@ -681,7 +703,7 @@ def send_audio( to: str | int, audio: str | pathlib.Path | bytes | BinaryIO, mime_type: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send an audio file to a WhatsApp user. @@ -699,7 +721,7 @@ def send_audio( audio: The audio file to send (either a media ID, URL, file path, bytes, or an open file object). mime_type: The mime type of the audio file (optional, required when sending an audio file as bytes or a file object, or file path that does not have an extension). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent audio file. @@ -711,12 +733,14 @@ def send_audio( media_type="audio", filename=None, ) - return self.api.send_media( + return self.api.send_message( to=str(to), - media_id_or_url=audio, - is_url=is_url, - media_type="audio", - tracker=tracker, + typ=MessageType.AUDIO.value, + msg=_get_media_msg( + media_id_or_url=audio, + is_url=is_url, + ), + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def send_sticker( @@ -724,7 +748,7 @@ def send_sticker( to: str | int, sticker: str | pathlib.Path | bytes | BinaryIO, mime_type: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send a sticker to a WhatsApp user. @@ -744,7 +768,7 @@ def send_sticker( sticker: The sticker to send (either a media ID, URL, file path, bytes, or an open file object). mime_type: The mime type of the sticker (optional, required when sending a sticker as bytes or a file object, or file path that does not have an extension). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent message. @@ -756,12 +780,14 @@ def send_sticker( filename=None, media_type="sticker", ) - return self.api.send_media( + return self.api.send_message( to=str(to), - media_id_or_url=sticker, - is_url=is_url, - media_type="sticker", - tracker=tracker, + typ=MessageType.STICKER.value, + msg=_get_media_msg( + media_id_or_url=sticker, + is_url=is_url, + ), + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def send_reaction( @@ -769,13 +795,17 @@ def send_reaction( to: str | int, emoji: str, message_id: str, + tracker: CallbackDataT | None = None, ) -> str: """ React to a message with an emoji. - You can react to incoming messages by using the :py:func:`~pywa.types.base_update.BaseUserUpdate.react` method on every update. - >>> msg.react('👍') + >>> wa = WhatsApp(...) + >>> @wa.on_message() + ... def message_handler(wa: WhatsApp, msg: Message): + ... msg.react('👍') Example: @@ -790,28 +820,34 @@ def send_reaction( to: The phone ID of the WhatsApp user. emoji: The emoji to react with. message_id: The message ID to react to. + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the reaction (You can't use this message id to remove the reaction or perform any other action on it. instead, use the message ID of the message you reacted to). """ - return self.api.send_reaction( + return self.api.send_message( to=str(to), - emoji=emoji, - message_id=message_id, + typ=MessageType.REACTION.value, + msg={"emoji": emoji, "message_id": message_id}, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def remove_reaction( self, to: str | int, message_id: str, + tracker: CallbackDataT | None = None, ) -> str: """ - Remove a reaction from a message. + Remove reaction from a message. - You can remove reactions from incoming messages by using the :py:func:`~pywa.types.base_update.BaseUserUpdate.unreact` method on every update. - >>> msg.unreact() + >>> wa = WhatsApp(...) + >>> @wa.on_message() + ... def message_handler(wa: WhatsApp, msg: Message): + ... msg.unreact() Example: @@ -824,14 +860,18 @@ def remove_reaction( Args: to: The phone ID of the WhatsApp user. message_id: The message ID to remove the reaction from. + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the reaction (You can't use this message id to re-react or perform any other action on it. instead, use the message ID of the message you unreacted to). """ - return self.api.send_reaction(to=str(to), emoji="", message_id=message_id)[ - "messages" - ][0]["id"] + return self.api.send_message( + to=str(to), + typ=MessageType.REACTION.value, + msg={"emoji": "", "message_id": message_id}, + tracker=_resolve_tracker_param(tracker), + )["messages"][0]["id"] def send_location( self, @@ -840,7 +880,7 @@ def send_location( longitude: float, name: str | None = None, address: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send a location to a WhatsApp user. @@ -862,22 +902,25 @@ def send_location( longitude: The longitude of the location. name: The name of the location (optional). address: The address of the location (optional). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent location. """ - return self.api.send_location( + return self.api.send_message( to=str(to), - latitude=latitude, - longitude=longitude, - name=name, - address=address, - tracker=tracker, + typ=MessageType.LOCATION.value, + msg={ + "latitude": latitude, + "longitude": longitude, + "name": name, + "address": address, + }, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def request_location( - self, to: str | int, text: str, tracker: str | None = None + self, to: str | int, text: str, tracker: CallbackDataT | None = None ) -> str: """ Send a text message with button to request the user's location. @@ -885,17 +928,20 @@ def request_location( Args: to: The phone ID of the WhatsApp user. text: The text to send with the button. - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent message. """ - return self.api.send_interactive_message( + return self.api.send_message( to=str(to), - type_="location_request_message", - action={"name": "send_location"}, - body=text, - tracker=tracker, + typ=MessageType.INTERACTIVE.value, + msg=_get_interactive_msg( + typ=InteractiveType.LOCATION_REQUEST_MESSAGE.value, + action={"name": "send_location"}, + body=text, + ), + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def send_contact( @@ -903,7 +949,7 @@ def send_contact( to: str | int, contact: Contact | Iterable[Contact], reply_to_message_id: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send a contact/s to a WhatsApp user. @@ -926,18 +972,19 @@ def send_contact( to: The phone ID of the WhatsApp user. contact: The contact/s to send. reply_to_message_id: The message ID to reply to (optional). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent message. """ - return self.api.send_contacts( + return self.api.send_message( to=str(to), - contacts=tuple(c.to_dict() for c in contact) + typ=MessageType.CONTACTS.value, + msg=tuple(c.to_dict() for c in contact) if isinstance(contact, Iterable) else (contact.to_dict(),), reply_to_message_id=reply_to_message_id, - tracker=tracker, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def send_catalog( @@ -947,7 +994,7 @@ def send_catalog( footer: str | None = None, thumbnail_product_sku: str | None = None, reply_to_message_id: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send the business catalog to a WhatsApp user. @@ -969,30 +1016,33 @@ def send_catalog( thumbnail_product_sku: The thumbnail of this item will be used as the message's header image (optional, if not provided, the first item in the catalog will be used). reply_to_message_id: The message ID to reply to (optional). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent message. """ - return self.api.send_interactive_message( + return self.api.send_message( to=str(to), - type_="catalog_message", - action={ - "name": "catalog_message", - **( - { - "parameters": { - "thumbnail_product_retailer_id": thumbnail_product_sku, + typ=MessageType.INTERACTIVE.value, + msg=_get_interactive_msg( + typ=InteractiveType.CATALOG_MESSAGE.value, + action={ + "name": "catalog_message", + **( + { + "parameters": { + "thumbnail_product_retailer_id": thumbnail_product_sku, + } } - } - if thumbnail_product_sku - else {} - ), - }, - body=body, - footer=footer, + if thumbnail_product_sku + else {} + ), + }, + body=body, + footer=footer, + ), reply_to_message_id=reply_to_message_id, - tracker=tracker, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def send_product( @@ -1003,7 +1053,7 @@ def send_product( body: str | None = None, footer: str | None = None, reply_to_message_id: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send a product from a business catalog to a WhatsApp user. @@ -1029,22 +1079,25 @@ def send_product( body: Text to appear in the message body (up to 1024 characters). footer: Text to appear in the footer of the message (optional, up to 60 characters). reply_to_message_id: The message ID to reply to (optional). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent message. """ - return self.api.send_interactive_message( + return self.api.send_message( to=str(to), - type_="product", - action={ - "catalog_id": catalog_id, - "product_retailer_id": sku, - }, - body=body, - footer=footer, + typ=MessageType.INTERACTIVE.value, + msg=_get_interactive_msg( + typ=InteractiveType.PRODUCT.value, + action={ + "catalog_id": catalog_id, + "product_retailer_id": sku, + }, + body=body, + footer=footer, + ), reply_to_message_id=reply_to_message_id, - tracker=tracker, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def send_products( @@ -1056,7 +1109,7 @@ def send_products( body: str, footer: str | None = None, reply_to_message_id: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send products from a business catalog to a WhatsApp user. @@ -1094,23 +1147,29 @@ def send_products( body: Text to appear in the message body (up to 1024 characters). footer: Text to appear in the footer of the message (optional, up to 60 characters). reply_to_message_id: The message ID to reply to (optional). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent message. """ - return self.api.send_interactive_message( + return self.api.send_message( to=str(to), - type_="product_list", - action={ - "catalog_id": catalog_id, - "sections": tuple(ps.to_dict() for ps in product_sections), - }, - header={"type": "text", "text": title}, - body=body, - footer=footer, + typ=MessageType.INTERACTIVE.value, + msg=_get_interactive_msg( + typ=InteractiveType.PRODUCT_LIST.value, + action={ + "catalog_id": catalog_id, + "sections": tuple(ps.to_dict() for ps in product_sections), + }, + header={ + "type": MessageType.TEXT.value, + "text": title, + }, + body=body, + footer=footer, + ), reply_to_message_id=reply_to_message_id, - tracker=tracker, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def mark_message_as_read( @@ -1524,7 +1583,7 @@ def send_template( to: str | int, template: Template, reply_to_message_id: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Send a template to a WhatsApp user. @@ -1571,7 +1630,7 @@ def send_template( to: The phone ID of the WhatsApp user. template: The template to send. reply_to_message_id: The message ID to reply to (optional). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The message ID of the sent template. @@ -1605,11 +1664,12 @@ def send_template( media_type="video", filename=None, ) - return self.api.send_template( + return self.api.send_message( to=str(to), - template=template.to_dict(is_header_url=is_url), + typ="template", + msg=template.to_dict(is_header_url=is_url), reply_to_message_id=reply_to_message_id, - tracker=tracker, + tracker=_resolve_tracker_param(tracker), )["messages"][0]["id"] def create_flow( @@ -1833,24 +1893,6 @@ def deprecate_flow( """ return self.api.deprecate_flow(flow_id=str(flow_id))["success"] - @staticmethod - def _get_flow_fields(invalidate_preview: bool) -> tuple[str, ...]: - """Internal method to get the fields of a flow.""" - return ( - "id", - "name", - "status", - "updated_at", - "categories", - "validation_errors", - "json_version", - "data_api_version", - "endpoint_uri", - f"preview.invalidate({'true' if invalidate_preview else 'false'})", - "whatsapp_business_account", - "application", - ) - def get_flow( self, flow_id: str | int, @@ -1869,7 +1911,7 @@ def get_flow( return FlowDetails.from_dict( data=self.api.get_flow( flow_id=str(flow_id), - fields=self._get_flow_fields(invalidate_preview=invalidate_preview), + fields=_get_flow_fields(invalidate_preview=invalidate_preview), ), client=self, ) @@ -1894,7 +1936,7 @@ def get_flows( FlowDetails.from_dict(data=data, client=self) for data in self.api.get_flows( business_account_id=self.business_account_id, - fields=self._get_flow_fields(invalidate_preview=invalidate_preview), + fields=_get_flow_fields(invalidate_preview=invalidate_preview), )["data"] ) @@ -1953,18 +1995,20 @@ def register_phone_number( def _resolve_buttons_param( buttons: Iterable[Button] | ButtonUrl | FlowButton | SectionList, -) -> tuple[str, dict]: +) -> tuple[InteractiveType, dict]: """ Internal method to resolve `buttons` parameter. Returns a tuple of (type, buttons). """ if isinstance(buttons, SectionList): - return "list", buttons.to_dict() + return InteractiveType.LIST.value, buttons.to_dict() elif isinstance(buttons, ButtonUrl): - return "cta_url", buttons.to_dict() + return InteractiveType.CTA_URL.value, buttons.to_dict() elif isinstance(buttons, FlowButton): - return "flow", buttons.to_dict() + return InteractiveType.FLOW.value, buttons.to_dict() else: # assume its a list of buttons - return "button", {"buttons": tuple(b.to_dict() for b in buttons)} + return InteractiveType.BUTTON.value, { + "buttons": tuple(b.to_dict() for b in buttons) + } _media_types_default_filenames = { @@ -1996,3 +2040,55 @@ def _resolve_media_param( mime_type=mime_type, filename=_media_types_default_filenames.get(media_type, filename), ) + + +def _resolve_tracker_param(tracker: CallbackDataT | None) -> str | None: + """Internal method to resolve the `tracker` parameter.""" + return tracker.to_str() if isinstance(tracker, CallbackData) else tracker + + +def _get_interactive_msg( + typ: InteractiveType, + action: dict[str, Any], + header: dict | None = None, + body: str | None = None, + footer: str | None = None, +): + return { + "type": typ, + "action": action, + **({"header": header} if header else {}), + **({"body": {"text": body}} if body else {}), + **({"footer": {"text": footer}} if footer else {}), + } + + +def _get_media_msg( + media_id_or_url: str, + is_url: bool, + caption: str | None = None, + filename: str | None = None, +): + return { + ("link" if is_url else "id"): media_id_or_url, + **({"caption": caption} if caption else {}), + **({"filename": filename} if filename else {}), + } + + +def _get_flow_fields(invalidate_preview: bool) -> tuple[str, ...]: + """Internal method to get the fields of a flow.""" + return ( + "id", + "name", + "status", + "updated_at", + "categories", + "validation_errors", + "json_version", + "data_api_version", + "endpoint_uri", + f"preview.invalidate({'true' if invalidate_preview else 'false'})", + "whatsapp_business_account", + "application", + ) diff --git a/pywa/handlers.py b/pywa/handlers.py index 66183fc..2147816 100644 --- a/pywa/handlers.py +++ b/pywa/handlers.py @@ -31,8 +31,8 @@ FlowRequest, FlowResponse, ChatOpened, + CallbackData, ) -from pywa.types.callback import CallbackData from pywa.types.flows import FlowCompletion, FlowResponseError # noqa if TYPE_CHECKING: @@ -59,6 +59,7 @@ def _safe_issubclass(obj: type, base: type) -> bool: def _resolve_callback_data_factory( factory: CallbackDataFactoryT, + field_name: str, ) -> tuple[CallbackDataFactoryT, Callable[[WhatsApp, Any], bool] | None]: """Internal function to resolve the callback data factory into a constractor and a filter.""" clb_filter = None @@ -84,9 +85,12 @@ def constractor(data: str) -> tuple[Any, ...] | list[Any, ...]: ): def clb_filter( - _: WhatsApp, btn_or_sel: CallbackButton | CallbackSelection + _: WhatsApp, update: CallbackButton | CallbackSelection | MessageStatus ) -> bool: - datas = btn_or_sel.data.split(CallbackData.__callback_sep__) + raw_data = getattr(update, field_name) + if raw_data is None: + return False + datas = raw_data.split(CallbackData.__callback_sep__) if len(datas) != len(factory): return False return all( @@ -104,11 +108,14 @@ def clb_filter( constractor = factory.from_str def clb_filter( - _: WhatsApp, btn_or_sel: CallbackButton | CallbackSelection + _: WhatsApp, update: CallbackButton | CallbackSelection | MessageStatus ) -> bool: + raw_data = getattr(update, field_name) + if raw_data is None: + return False return len( - btn_or_sel.data.split(CallbackData.__callback_sep__) - ) == 1 and btn_or_sel.data.startswith( + raw_data.split(CallbackData.__callback_sep__) + ) == 1 and raw_data.startswith( str(factory.__callback_id__) + factory.__callback_data_sep__ ) @@ -119,22 +126,24 @@ def clb_filter( def _call_callback_handler( - handler: CallbackButtonHandler | CallbackSelectionHandler, + handler: CallbackButtonHandler | CallbackSelectionHandler | MessageStatusHandler, wa: WhatsApp, - clb_or_sel: CallbackButton | CallbackSelection, + update: CallbackButton | CallbackSelection | MessageStatus, + field_name: str, ): """Internal function to call a callback callback.""" if handler.factory_before_filters and handler.factory_filter is not None: - if handler.factory_filter(wa, clb_or_sel): - clb_or_sel = dataclasses.replace( - clb_or_sel, data=handler.factory(clb_or_sel.data) + if handler.factory_filter(wa, update): + data = getattr(update, field_name) + update = dataclasses.replace( + update, data=handler.factory(data) if data else None ) else: return try: pass_filters = all( ( - f(wa, clb_or_sel) + f(wa, update) for f in ( *( (handler.factory_filter,) @@ -159,10 +168,11 @@ def _call_callback_handler( raise if pass_filters: if not handler.factory_before_filters and handler.factory is not str: - clb_or_sel = dataclasses.replace( - clb_or_sel, data=handler.factory(clb_or_sel.data) + data = getattr(update, field_name) + update = dataclasses.replace( + update, **{field_name: handler.factory(data) if data else None} ) - handler.callback(wa, clb_or_sel) + handler.callback(wa, update) class Handler(abc.ABC): @@ -274,7 +284,7 @@ class CallbackButtonHandler(Handler): *filters: The filters to apply to the handler (Takes a :class:`pywa.WhatsApp` instance and a :class:`pywa.types.CallbackButton` and returns a :class:`bool`) factory: The constructor/s to use to construct the callback data (default: :class:`str`. If the factory is a - subclass of :class:`pywa.types.CallbackData`, a matching filter is automatically added). + subclass of :class:`CallbackData`, a matching filter is automatically added). factory_before_filters: Whether to apply the factory before the filters (default: ``False``. If ``True``, the filters will get the callback data after the factory is applied). """ @@ -292,12 +302,12 @@ def __init__( ( self.factory, self.factory_filter, - ) = _resolve_callback_data_factory(factory) + ) = _resolve_callback_data_factory(factory, "data") self.factory_before_filters = factory_before_filters super().__init__(callback, *filters) def handle(self, wa: WhatsApp, clb: CallbackButton): - _call_callback_handler(self, wa, clb) + _call_callback_handler(self, wa, clb, "data") def __str__(self) -> str: return f"{self.__class__.__name__}(callback={self.callback!r}, filters={self.filters!r}, factory={self.factory!r})" @@ -322,7 +332,7 @@ class CallbackSelectionHandler(Handler): *filters: The filters to apply to the handler. (Takes a :class:`pywa.WhatsApp` instance and a :class:`pywa.types.CallbackSelection` and returns a :class:`bool`) factory: The constructor/s to use to construct the callback data (default: :class:`str`. If the factory is a - subclass of :class:`pywa.types.CallbackData`, a matching filter is automatically added). + subclass of :class:`CallbackData`, a matching filter is automatically added). factory_before_filters: Whether to apply the factory before the filters (default: ``False``. If ``True``, the filters will get the callback data after the factory is applied). """ @@ -340,12 +350,12 @@ def __init__( ( self.factory, self.factory_filter, - ) = _resolve_callback_data_factory(factory) + ) = _resolve_callback_data_factory(factory, "data") self.factory_before_filters = factory_before_filters super().__init__(callback, *filters) def handle(self, wa: WhatsApp, sel: CallbackSelection): - _call_callback_handler(self, wa, sel) + _call_callback_handler(self, wa, sel, "data") def __str__(self) -> str: return f"{self.__class__.__name__}(callback={self.callback!r}, filters={self.filters!r}, factory={self.factory!r})" @@ -371,6 +381,10 @@ class MessageStatusHandler(Handler): arguments) *filters: The filters to apply to the handler (Takes a :class:`pywa.WhatsApp` instance and a :class:`pywa.types.MessageStatus` and returns a :class:`bool`) + factory: The constructor/s to use to construct the tracker data (default: :class:`str`. If the factory is a + subclass of :class:`CallbackData`, a matching filter is automatically added). + factory_before_filters: Whether to apply the factory before the filters (default: ``False``. If ``True``, the + filters will get the tracker data after the factory is applied). """ _field_name = "messages" @@ -380,9 +394,22 @@ def __init__( self, callback: Callable[[WhatsApp, MessageStatus], Any], *filters: Callable[[WhatsApp, MessageStatus], bool], + factory: CallbackDataFactoryT = str, + factory_before_filters: bool = False, ): + ( + self.factory, + self.factory_filter, + ) = _resolve_callback_data_factory(factory, "tracker") + self.factory_before_filters = factory_before_filters super().__init__(callback, *filters) + def handle(self, wa: WhatsApp, status: MessageStatus): + _call_callback_handler(self, wa, status, "tracker") + + def __str__(self) -> str: + return f"{self.__class__.__name__}(callback={self.callback!r}, filters={self.filters!r}, factory={self.factory!r})" + class ChatOpenedHandler(Handler): """ @@ -654,7 +681,7 @@ def on_callback_button( *filters: Filters to apply to the incoming callback button presses (filters are function that take a :class:`pywa.WhatsApp` instance and the incoming :class:`pywa.types.CallbackButton` and return :class:`bool`). factory: The constructor/s to use for the callback data (default: :class:`str`. If the factory is a - subclass of :class:`pywa.types.CallbackData`, a matching filter is automatically added). + subclass of :class:`CallbackData`, a matching filter is automatically added). factory_before_filters: Whether to apply the factory before the filters (default: ``False``. If ``True``, the filters will get the callback data after the factory is applied). """ @@ -702,7 +729,7 @@ def on_callback_selection( *filters: Filters to apply to the incoming callback selections (filters are function that take a :class:`pywa.WhatsApp` instance and the incoming :class:`pywa.types.CallbackSelection` and return :class:`bool`). factory: The constructor/s to use for the callback data (default: :class:`str`. If the factory is a - subclass of :class:`pywa.types.CallbackData`, a matching filter is automatically added). + subclass of :class:`CallbackData`, a matching filter is automatically added). factory_before_filters: Whether to apply the factory before the filters (default: ``False``. If ``True``, the filters will get the callback data after the factory is applied). """ @@ -726,6 +753,8 @@ def decorator( def on_message_status( self: WhatsApp, *filters: Callable[[WhatsApp, MessageStatus], bool], + factory: CallbackDataFactoryT = str, + factory_before_filters: bool = False, ) -> Callable[ [Callable[[WhatsApp, MessageStatus], Any]], Callable[[WhatsApp, MessageStatus], Any], @@ -751,13 +780,24 @@ def on_message_status( Args: *filters: Filters to apply to the incoming message status changes (filters are function that take a :class:`pywa.WhatsApp` instance and the incoming :class:`pywa.types.MessageStatus` and return :class:`bool`). + factory: The constructor/s to use for the tracker data (default: :class:`str`. If the factory is a + subclass of :class:`CallbackData`, a matching filter is automatically added). + factory_before_filters: Whether to apply the factory before the filters (default: ``False``. If ``True``, the + filters will get the tracker data after the factory is applied). """ @functools.wraps(self.on_message_status) def decorator( callback: Callable[[WhatsApp, MessageStatus], Any], ) -> Callable[[WhatsApp, MessageStatus], Any]: - self.add_handlers(MessageStatusHandler(callback, *filters)) + self.add_handlers( + MessageStatusHandler( + callback, + *filters, + factory=factory, + factory_before_filters=factory_before_filters, + ) + ) return callback return decorator diff --git a/pywa/types/base_update.py b/pywa/types/base_update.py index d3f2c5d..d6aa307 100644 --- a/pywa/types/base_update.py +++ b/pywa/types/base_update.py @@ -10,7 +10,7 @@ import abc import pathlib import dataclasses -import datetime as dt +import datetime from typing import TYPE_CHECKING, BinaryIO, Iterable from .others import Contact, Metadata, ProductsSection, User @@ -18,7 +18,7 @@ if TYPE_CHECKING: from pywa.client import WhatsApp - from .callback import Button, ButtonUrl, SectionList, FlowButton + from .callback import Button, ButtonUrl, SectionList, FlowButton, CallbackDataT from .template import Template @@ -62,7 +62,7 @@ def id(self) -> str: @property @abc.abstractmethod - def timestamp(self) -> dt.datetime: + def timestamp(self) -> datetime.datetime: """The timestamp the update was sent""" ... @@ -139,7 +139,7 @@ def reply_text( quote: bool = False, preview_url: bool = False, keyboard: None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with text. @@ -216,7 +216,7 @@ def reply_text( quote: Whether to quote the replied message (default: False). preview_url: Whether to show a preview of the URL in the message (if any). keyboard: Deprecated and will be removed in a future version, use ``buttons`` instead. - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. @@ -244,7 +244,7 @@ def reply_image( buttons: Iterable[Button] | ButtonUrl | FlowButton | None = None, quote: bool = False, mime_type: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with an image. @@ -271,7 +271,7 @@ def reply_image( or file path that does not have an extension). quote: Whether to quote the replied message (default: False). body: Deprecated and will be removed in a future version, use ``caption`` instead. - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. @@ -297,7 +297,7 @@ def reply_video( buttons: Iterable[Button] | ButtonUrl | FlowButton | None = None, quote: bool = False, mime_type: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with a video. @@ -325,7 +325,7 @@ def reply_video( or file path that does not have an extension). quote: Whether to quote the replied message (default: False). body: Deprecated and will be removed in a future version, use ``caption`` instead. - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. @@ -339,6 +339,7 @@ def reply_video( body=body, footer=footer, mime_type=mime_type, + tracker=tracker, ) def reply_document( @@ -351,7 +352,7 @@ def reply_document( buttons: Iterable[Button] | ButtonUrl | FlowButton | None = None, quote: bool = False, mime_type: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with a document. @@ -381,7 +382,7 @@ def reply_document( object, or file path that does not have an extension). body: Deprecated and will be removed in a future version, use ``caption`` instead. quote: Whether to quote the replied message (default: False). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. @@ -403,7 +404,7 @@ def reply_audio( self, audio: str | pathlib.Path | bytes | BinaryIO, mime_type: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with an audio. @@ -419,7 +420,7 @@ def reply_audio( audio: The audio file to reply with (either a media ID, URL, file path, bytes, or an open file object). mime_type: The mime type of the audio (optional, required when sending a audio as bytes or a file object, or file path that does not have an extension). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent message. @@ -435,7 +436,7 @@ def reply_sticker( self, sticker: str | pathlib.Path | bytes | BinaryIO, mime_type: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with a sticker. @@ -453,7 +454,7 @@ def reply_sticker( sticker: The sticker to reply with (either a media ID, URL, file path, bytes, or an open file object). mime_type: The mime type of the sticker (optional, required when sending a sticker as bytes or a file object, or file path that does not have an extension). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. @@ -471,7 +472,7 @@ def reply_location( longitude: float, name: str | None = None, address: str | None = None, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with a location. @@ -492,7 +493,7 @@ def reply_location( longitude: The longitude of the location. name: The name of the location (optional). address: The address of the location (optional). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. @@ -510,7 +511,7 @@ def reply_contact( self, contact: Contact | Iterable[Contact], quote: bool = False, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with a contact/s. @@ -533,7 +534,7 @@ def reply_contact( Args: contact: The contact/s to send. quote: Whether to quote the replied message (default: False). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. @@ -545,7 +546,7 @@ def reply_contact( tracker=tracker, ) - def react(self, emoji: str) -> str: + def react(self, emoji: str, tracker: CallbackDataT | None = None) -> str: """ React to the message with an emoji. - Shortcut for :py:func:`~pywa.client.WhatsApp.send_reaction` with ``to`` and ``message_id``. @@ -556,6 +557,7 @@ def react(self, emoji: str) -> str: Args: emoji: The emoji to react with. + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reaction. @@ -564,9 +566,10 @@ def react(self, emoji: str) -> str: to=self.sender, emoji=emoji, message_id=self.message_id_to_reply, + tracker=tracker, ) - def unreact(self) -> str: + def unreact(self, tracker: CallbackDataT | None = None) -> str: """ Remove the reaction from the message. - Shortcut for :py:func:`~pywa.client.WhatsApp.remove_reaction` with ``to`` and ``message_id``. @@ -575,11 +578,14 @@ def unreact(self) -> str: >>> msg.unreact() + Args: + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). + Returns: The ID of the sent unreaction. """ return self._client.remove_reaction( - to=self.sender, message_id=self.message_id_to_reply + to=self.sender, message_id=self.message_id_to_reply, tracker=tracker ) def reply_catalog( @@ -588,7 +594,7 @@ def reply_catalog( footer: str | None = None, thumbnail_product_sku: str | None = None, quote: bool = False, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with a catalog. @@ -608,7 +614,7 @@ def reply_catalog( thumbnail_product_sku: The thumbnail of this item will be used as the message's header image (optional, if not provided, the first item in the catalog will be used). quote: Whether to quote the replied message (default: False). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. @@ -629,7 +635,7 @@ def reply_product( body: str | None = None, footer: str | None = None, quote: bool = False, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with a product. @@ -644,7 +650,7 @@ def reply_product( body: Text to appear in the message body (up to 1024 characters). footer: Text to appear in the footer of the message (optional, up to 60 characters). quote: Whether to quote the replied message (default: False). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. @@ -667,7 +673,7 @@ def reply_products( body: str, footer: str | None = None, quote: bool = False, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with a product. @@ -704,7 +710,7 @@ def reply_products( body: Text to appear in the message body (up to 1024 characters). footer: Text to appear in the footer of the message (optional, up to 60 characters). quote: Whether to quote the replied message (default: False). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. @@ -724,7 +730,7 @@ def reply_template( self, template: Template, quote: bool = False, - tracker: str | None = None, + tracker: CallbackDataT | None = None, ) -> str: """ Reply to the message with a template. @@ -769,7 +775,7 @@ def reply_template( Args: template: The template to send. quote: Whether to quote the replied message (default: False). - tracker: The data to track the message with (optional, up to 512 characters). + tracker: The data to track the message with (optional, up to 512 characters, for complex data You can use :class:`CallbackData`). Returns: The ID of the sent reply. diff --git a/pywa/types/callback.py b/pywa/types/callback.py index b32ce28..7d3731b 100644 --- a/pywa/types/callback.py +++ b/pywa/types/callback.py @@ -14,7 +14,7 @@ ] import dataclasses -import datetime as dt +import datetime import enum import types from typing import ( @@ -32,7 +32,7 @@ from .base_update import BaseUserUpdate # noqa from .flows import FlowStatus, FlowActionType -from .others import MessageType, Metadata, ReplyToMessage, User +from .others import MessageType, Metadata, ReplyToMessage, User, InteractiveType from .. import utils if TYPE_CHECKING: @@ -244,7 +244,7 @@ def join_to_str(cls, *datas: Any) -> str: "CallbackDataT", bound=str | CallbackData | Iterable[CallbackData | Any], ) -"""Type hint for ``callback_data`` parameter in :class:`Button` and :class:`SectionRow`.""" +"""Type hint for ``callback_data`` parameter in :class:`Button` and :class:`SectionRow` and for ``tracker``'s""" @dataclasses.dataclass(frozen=True, slots=True, kw_only=True) @@ -316,7 +316,7 @@ class CallbackButton(BaseUserUpdate, Generic[CallbackDataT]): type: MessageType metadata: Metadata from_user: User - timestamp: dt.datetime + timestamp: datetime.datetime reply_to_message: ReplyToMessage data: CallbackDataT title: str @@ -340,7 +340,7 @@ def from_update(cls, client: "WhatsApp", update: dict) -> "CallbackButton": metadata=Metadata.from_dict(value["metadata"]), type=MessageType(msg_type), from_user=User.from_dict(value["contacts"][0]), - timestamp=dt.datetime.fromtimestamp(int(msg["timestamp"])), + timestamp=datetime.datetime.fromtimestamp(int(msg["timestamp"])), reply_to_message=ReplyToMessage.from_dict(msg["context"]), data=data, title=title, @@ -424,7 +424,7 @@ class CallbackSelection(BaseUserUpdate, Generic[CallbackDataT]): type: MessageType metadata: Metadata from_user: User - timestamp: dt.datetime + timestamp: datetime.datetime reply_to_message: ReplyToMessage data: CallbackDataT title: str @@ -440,7 +440,7 @@ def from_update(cls, client: "WhatsApp", update: dict) -> "CallbackSelection": metadata=Metadata.from_dict(value["metadata"]), type=MessageType(msg["type"]), from_user=User.from_dict(value["contacts"][0]), - timestamp=dt.datetime.fromtimestamp(int(msg["timestamp"])), + timestamp=datetime.datetime.fromtimestamp(int(msg["timestamp"])), reply_to_message=ReplyToMessage.from_dict(msg["context"]), data=msg["interactive"]["list_reply"]["id"], title=msg["interactive"]["list_reply"]["title"], @@ -499,7 +499,7 @@ class ButtonUrl: def to_dict(self) -> dict: return { - "name": "cta_url", + "name": InteractiveType.CTA_URL, "parameters": {"display_text": self.title, "url": self.url}, } @@ -613,7 +613,7 @@ def __post_init__(self): def to_dict(self) -> dict: return { - "name": "flow", + "name": InteractiveType.FLOW, "parameters": { "mode": self.mode.lower(), "flow_message_version": str(self.flow_message_version), diff --git a/pywa/types/message.py b/pywa/types/message.py index 9ac35b1..268b641 100644 --- a/pywa/types/message.py +++ b/pywa/types/message.py @@ -5,7 +5,7 @@ __all__ = ["Message"] import dataclasses -import datetime as dt +import datetime from typing import TYPE_CHECKING, Any, Callable, Iterable from pywa.errors import WhatsAppError @@ -84,7 +84,7 @@ class Message(BaseUserUpdate): type: MessageType metadata: Metadata from_user: User - timestamp: dt.datetime + timestamp: datetime.datetime reply_to_message: ReplyToMessage | None forwarded: bool forwarded_many_times: bool @@ -152,7 +152,7 @@ def from_update(cls, client: WhatsApp, update: dict) -> Message: type=MessageType(msg_type), **msg_content, from_user=usr, - timestamp=dt.datetime.fromtimestamp(int(msg["timestamp"])), + timestamp=datetime.datetime.fromtimestamp(int(msg["timestamp"])), metadata=Metadata.from_dict(value["metadata"]), forwarded=context.get("forwarded", False) or context.get("frequently_forwarded", False), diff --git a/pywa/types/message_status.py b/pywa/types/message_status.py index c221b93..fd620b6 100644 --- a/pywa/types/message_status.py +++ b/pywa/types/message_status.py @@ -1,5 +1,7 @@ from __future__ import annotations +from .callback import CallbackDataT + """This module contains the types related to message status updates.""" __all__ = [ @@ -11,8 +13,8 @@ import dataclasses import logging -import datetime as dt -from typing import TYPE_CHECKING +import datetime +from typing import TYPE_CHECKING, Generic from pywa import utils from pywa.errors import WhatsAppError @@ -34,7 +36,7 @@ class MessageStatusType(utils.StrEnum): SENT: The message was sent. DELIVERED: The message was delivered. READ: The message was read. - FAILED: The message failed to send (you can access the error with ``.error`` attribute). + FAILED: The message failed to send (you can access the ``.error`` attribute for more information). """ SENT = "sent" @@ -77,30 +79,75 @@ def _missing_(cls, value: str) -> ConversationCategory: @dataclasses.dataclass(frozen=True, slots=True, kw_only=True) -class MessageStatus(BaseUserUpdate): +class MessageStatus(BaseUserUpdate, Generic[CallbackDataT]): """ Represents the status of a message. - `'MessageStatus' on developers.facebook.com `_. + ``MessageStatus`` is a generic class, so when providing a ``factory`` parameter in callback handlers, you can + specify the type of the factory to get autocomplete in the ``tracker`` attribute. + + Here is an example: + + >>> from pywa.types import CallbackData + >>> from dataclasses import dataclass + >>> @dataclass(frozen=True, slots=True) + >>> class UserData(CallbackData): # Subclass CallbackData + ... id: int + ... name: str + ... admin: bool + + >>> from pywa import WhatsApp + >>> from pywa.types import Button, CallbackButton + >>> wa = WhatsApp(...) + >>> wa.send_message( + ... to='972987654321', + ... text='Hi user', + ... tracker=UserData(id=123, name='david', admin=True) # Here ^^^ we use the UserData class as the tracker + ... ) # Here ^^^ we use the UserData class as the tracker data + + >>> @wa.on_message_status(factory=UserData) # Use the factory parameter to convert the tracker data + ... def on_status(_: WhatsApp, s: MessageStatus[UserData]): # For autocomplete + ... if s.tracker.admin: print(s.tracker.id) # Access the tracker data + + You can even use multiple factories, and not only ``CallbackData`` subclasses! + + >>> from enum import Enum + >>> class State(str, Enum): + ... START = 's' + ... END = 'e' + + >>> wa.send_message( + ... to='972987654321', + ... text='Hi user', + ... tracker=(UserData(id=123, name='david', admin=True), State.START) + ... ) # Here ^^^ we send a tuple of UserData and State + + >>> @wa.on_message_status(factory=tuple[UserData, State]) # Use the factory parameter to convert the tracker data + ... def on_user_data(_: WhatsApp, s: MessageStatus[tuple[UserData, State]]): # For autocomplete + ... user, state = s.tracker # Unpack the tuple + ... if user.admin: print(user.id, state) + + Attributes: id: The ID of the message that the status is for. metadata: The metadata of the message (to which phone number it was sent). status: The status of the message. timestamp: The timestamp when the status was updated. from_user: The user who the message was sent to. - tracker: The tracker that the message was sent with. + tracker: The tracker that the message was sent with (e.g. ``wa.send_message(tracker=...)``). conversation: The conversation the given status notification belongs to (Optional). pricing_model: Type of pricing model used by the business. Current supported value is CBP. - error: The error that occurred (if status is ``failed``). + error: The error that occurred (if status is :class:`MessageStatusType.FAILED`). """ id: str metadata: Metadata from_user: User - timestamp: dt.datetime + timestamp: datetime.datetime status: MessageStatusType - tracker: str | None + tracker: CallbackDataT | None conversation: Conversation | None pricing_model: str | None error: WhatsAppError | None @@ -115,7 +162,7 @@ def from_update(cls, client: WhatsApp, update: dict) -> MessageStatus: id=status["id"], metadata=Metadata.from_dict(value["metadata"]), status=MessageStatusType(status["status"]), - timestamp=dt.datetime.fromtimestamp(int(status["timestamp"])), + timestamp=datetime.datetime.fromtimestamp(int(status["timestamp"])), from_user=User(wa_id=status["recipient_id"], name=None), tracker=status.get("biz_opaque_callback_data"), conversation=Conversation.from_dict(status["conversation"]) @@ -142,14 +189,16 @@ class Conversation: id: str category: ConversationCategory - expiration: dt.datetime | None + expiration: datetime.datetime | None @classmethod def from_dict(cls, data: dict): return cls( id=data["id"], category=ConversationCategory(data["origin"]["type"]), - expiration=dt.datetime.fromtimestamp(int(data["expiration_timestamp"])) + expiration=datetime.datetime.fromtimestamp( + int(data["expiration_timestamp"]) + ) if "expiration_timestamp" in data else None, ) diff --git a/pywa/types/others.py b/pywa/types/others.py index 4477fc2..38a7064 100644 --- a/pywa/types/others.py +++ b/pywa/types/others.py @@ -95,6 +95,24 @@ def _missing_(cls, value: str) -> MessageType: return cls.UNKNOWN +class InteractiveType(utils.StrEnum): + """ + Interactive types. + + Attributes: + + """ + + BUTTON = "button" + CTA_URL = "cta_url" + CATALOG_MESSAGE = "catalog_message" + LIST = "list" + PRODUCT = "product" + PRODUCT_LIST = "product_list" + FLOW = "flow" + LOCATION_REQUEST_MESSAGE = "location_request_message" + + @dataclasses.dataclass(frozen=True, slots=True) class Reaction(utils.FromDict): """ diff --git a/pywa/types/template.py b/pywa/types/template.py index 285a5f8..70b2d38 100644 --- a/pywa/types/template.py +++ b/pywa/types/template.py @@ -16,7 +16,7 @@ import logging import re import pathlib -import datetime as dt +import datetime from typing import TYPE_CHECKING, Any, BinaryIO, Iterable, Literal from pywa import utils @@ -1437,7 +1437,7 @@ class TemplateStatus(BaseUpdate): """ id: str - timestamp: dt.datetime + timestamp: datetime.datetime event: TemplateEvent message_template_id: int message_template_name: str @@ -1453,7 +1453,7 @@ def from_update(cls, client: WhatsApp, update: dict) -> TemplateStatus: _client=client, raw=update, id=data["id"], - timestamp=dt.datetime.fromtimestamp(data["time"]), + timestamp=datetime.datetime.fromtimestamp(data["time"]), event=cls.TemplateEvent(value["event"]), message_template_id=value["message_template_id"], message_template_name=value["message_template_name"],