Skip to content

Commit

Permalink
[api, client] Added support for block_users and unblock_users
Browse files Browse the repository at this point in the history
  • Loading branch information
yehuda-lev committed Jan 19, 2025
1 parent 90e2419 commit af09c99
Show file tree
Hide file tree
Showing 4 changed files with 188 additions and 1 deletion.
54 changes: 54 additions & 0 deletions pywa/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 <https://developers.facebook.com/docs/whatsapp/cloud-api/reference/block-users>`_.
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 <https://developers.facebook.com/docs/whatsapp/cloud-api/reference/block-users>`_.
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,
Expand Down
44 changes: 44 additions & 0 deletions pywa/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
44 changes: 44 additions & 0 deletions pywa_async/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 <https://developers.facebook.com/docs/whatsapp/cloud-api/block-users>`_.
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 <https://developers.facebook.com/docs/whatsapp/cloud-api/block-users>`_.
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,
Expand Down
47 changes: 46 additions & 1 deletion pywa_async/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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,
Expand Down

0 comments on commit af09c99

Please sign in to comment.