From af09c99d3a4040aaa3a6ebd17a8a6ccb2418c7c1 Mon Sep 17 00:00:00 2001 From: yehuda-lev <94082865+yehuda-lev@users.noreply.github.com> Date: Sun, 19 Jan 2025 23:52:29 +0200 Subject: [PATCH] [api, client] Added support for `block_users` and `unblock_users` --- pywa/api.py | 54 ++++++++++++++++++++++++++++++++++++++++++++ pywa/client.py | 44 ++++++++++++++++++++++++++++++++++++ pywa_async/api.py | 44 ++++++++++++++++++++++++++++++++++++ pywa_async/client.py | 47 +++++++++++++++++++++++++++++++++++++- 4 files changed, 188 insertions(+), 1 deletion(-) diff --git a/pywa/api.py b/pywa/api.py index 4b438fd..f0ed59b 100644 --- a/pywa/api.py +++ b/pywa/api.py @@ -567,6 +567,60 @@ def mark_message_as_read( }, ) + def block_users( + self, + phone_id: str, + users: list[str], + ) -> dict[str, Any]: + """ + Block users from sending messages to the business. + + - Read more at `developers.facebook.com `_. + + + Args: + phone_id: The ID of the phone number to block users on. + users: The list of users to block. + + Returns: + The response from the WhatsApp Cloud API. + """ + return self._make_request( + method="POST", + endpoint=f"/{phone_id}/block_users", + json={ + "messaging_product": "whatsapp", + "block_users": [{"user": user} for user in users], + }, + ) + + def unblock_users( + self, + phone_id: str, + users: list[str], + ) -> dict[str, Any]: + """ + Unblock users from sending messages to the business. + + - Read more at `developers.facebook.com `_. + + + Args: + phone_id: The ID of the phone number to unblock users on. + users: The list of users to unblock. + + Returns: + The response from the WhatsApp Cloud API. + """ + return self._make_request( + method="DELETE", + endpoint=f"/{phone_id}/block_users", + json={ + "messaging_product": "whatsapp", + "block_users": [{"user": user} for user in users], + }, + ) + def get_business_phone_number( self, phone_id: str, diff --git a/pywa/client.py b/pywa/client.py index 14e0d7b..dc7c640 100644 --- a/pywa/client.py +++ b/pywa/client.py @@ -1428,6 +1428,50 @@ def mark_message_as_read( message_id=message_id, )["success"] + def block_users( + self, users: Iterable[str | int], phone_id: str | int | None = None + ) -> dict[str, Any]: + """ + Block users by phone ID. + + Example: + + >>> wa = WhatsApp(...) + >>> wa.block_users(users=['1234567890', '0987654321']) + + Args: + users: The phone IDs of the users to block. + phone_id: The phone ID to block the users from (optional, if not provided, the client's phone ID will be used). + Returns: + A dictionary with the status of the block operation. + """ + return self.api.block_users( + phone_id=helpers.resolve_phone_id_param(self, phone_id, "phone_id"), + users=[str(phone_id) for phone_id in users], + ) + + def unblock_users( + self, users: Iterable[str | int], phone_id: str | int | None = None + ) -> dict[str, Any]: + """ + Unblock users by phone ID. + + Example: + + >>> wa = WhatsApp(...) + >>> wa.unblock_users(users=['1234567890', '0987654321']) + + Args: + users: The phone IDs of the users to unblock. + phone_id: The phone ID to unblock the users from (optional, if not provided, the client's phone ID will be used). + Returns: + A dictionary with the status of the unblock operation. + """ + return self.api.unblock_users( + phone_id=helpers.resolve_phone_id_param(self, phone_id, "phone_id"), + users=[str(phone_id) for phone_id in users], + ) + def upload_media( self, media: str | pathlib.Path | bytes | BinaryIO, diff --git a/pywa_async/api.py b/pywa_async/api.py index 79c75cd..45f50e3 100644 --- a/pywa_async/api.py +++ b/pywa_async/api.py @@ -557,6 +557,50 @@ async def mark_message_as_read( }, ) + async def block_users(self, phone_id: str, users: list[str]) -> dict[str, Any]: + """ + Block users on a phone number. + + - Read more at `developers.facebook.com `_. + + Args: + phone_id: The ID of the phone number to block users on. + users: The list of users to block. + + Returns: + The response from the WhatsApp Cloud API. + """ + return await self._make_request( + method="POST", + endpoint=f"/{phone_id}/block_users", + json={ + "messaging_product": "whatsapp", + "block_users": [{"user": user} for user in users], + }, + ) + + async def unblock_users(self, phone_id: str, users: list[str]) -> dict[str, Any]: + """ + Unblock users on a phone number. + + - Read more at `developers.facebook.com `_. + + Args: + phone_id: The ID of the phone number to unblock users on. + users: The list of users to unblock. + + Returns: + The response from the WhatsApp Cloud API. + """ + return await self._make_request( + method="DELETE", + endpoint=f"/{phone_id}/block_users", + json={ + "messaging_product": "whatsapp", + "block_users": [{"user": user} for user in users], + }, + ) + async def get_business_phone_number( self, phone_id: str, diff --git a/pywa_async/client.py b/pywa_async/client.py index 03d89d2..b7e2acb 100644 --- a/pywa_async/client.py +++ b/pywa_async/client.py @@ -14,7 +14,7 @@ import pathlib import warnings from types import ModuleType -from typing import BinaryIO, Iterable, Literal +from typing import BinaryIO, Iterable, Literal, Any import httpx @@ -1191,6 +1191,51 @@ async def mark_message_as_read( ) )["success"] + async def block_users( + self, users: Iterable[str | int], phone_id: str | int | None = None + ) -> dict[str, Any]: + """ + Block users by phone ID. + + Example: + + >>> wa = WhatsApp(...) + >>> wa.block_users(users=['1234567890', '0987654321']) + + Args: + users: The phone IDs of the users to block. + phone_id: The phone ID to block the users from (optional, if not provided, the client's phone ID will be used). + Returns: + A dictionary with the status of the block operation. + """ + + return await self.api.block_users( + phone_id=helpers.resolve_phone_id_param(self, phone_id, "phone_id"), + users=[str(phone_id) for phone_id in users], + ) + + async def unblock_users( + self, users: Iterable[str | int], phone_id: str | int | None = None + ) -> dict[str, Any]: + """ + Unblock users by phone ID. + + Example: + + >>> wa = WhatsApp(...) + >>> wa.unblock_users(users=['1234567890', '0987654321']) + + Args: + users: The phone IDs of the users to unblock. + phone_id: The phone ID to unblock the users from (optional, if not provided, the client's phone ID will be used). + Returns: + A dictionary with the status of the unblock operation. + """ + return await self.api.unblock_users( + phone_id=helpers.resolve_phone_id_param(self, phone_id, "phone_id"), + users=[str(phone_id) for phone_id in users], + ) + async def upload_media( self, media: str | pathlib.Path | bytes | BinaryIO,