# Wasender API Python SDK
**Python SDK Author:** YonkoSam
**Original Node.js SDK Author:** Shreshth Arora
[](https://pypi.org/project/wasenderapi/)
[](https://pypi.org/project/wasenderapi/)
[](LICENSE)
[](https://www.python.org/)
[](https://github.com/YonkoSam/wasenderapi-python/actions/workflows/ci.yml)
A lightweight and robust Python SDK for interacting with the Wasender API ([https://www.wasenderapi.com](https://www.wasenderapi.com)). This SDK simplifies sending various types of WhatsApp messages, managing contacts and groups, handling session statuses, and processing incoming webhooks.
## Features
- **Pydantic Models:** Leverages Pydantic for robust request/response validation and serialization, especially for the generic `send()` method and webhook event parsing.
- **Message Sending:**
- Simplified helper methods (e.g., `client.send_text(to="...", text_body="...")`, `client.send_image(to="...", url="...", caption="...")`) that accept direct parameters for common message types.
- Generic `client.send(payload: BaseMessage)` method for advanced use cases or less common message types, accepting a Pydantic model.
- Support for text, image, video, document, audio, sticker, contact card, and location messages.
- **Contact Management:** List, retrieve details, get profile pictures, block, and unblock contacts.
- **Group Management:** List groups, fetch metadata, manage participants (add/remove), and update group settings.
- **Channel Messaging:** Send text messages to WhatsApp Channels.
- **Session Management:** Create, list, update, delete sessions, connect/disconnect, get QR codes, and check session status.
- **Webhook Handling:** Securely verify and parse incoming webhook events from Wasender using Pydantic models.
- **Error Handling:** Comprehensive `WasenderAPIError` class with detailed error information.
- **Rate Limiting:** Access to rate limit information on API responses.
- **Retry Mechanism:** Optional automatic retries for rate-limited requests (HTTP 429) via `RetryConfig`.
- **Customizable HTTP Client:** Allows providing a custom `httpx.AsyncClient` instance for the asynchronous client.
## Prerequisites
- Python (version 3.8 or higher recommended).
- A Wasender API Key from [https://www.wasenderapi.com](https://www.wasenderapi.com).
- If using webhooks:
- A publicly accessible HTTPS URL for your webhook endpoint.
- A Webhook Secret generated from the Wasender dashboard.
## Installation
```bash
pip install wasenderapi
```
## SDK Initialization
The SDK now provides both a synchronous and an asynchronous client.
### Synchronous Client
```python
import os
from wasenderapi import WasenderSyncClient, create_sync_wasender
from wasenderapi.models import RetryConfig
# Required credentials
api_key = os.getenv("WASENDER_API_KEY")
# For account-scoped endpoints like session management:
personal_access_token = os.getenv("WASENDER_PERSONAL_ACCESS_TOKEN")
# For webhook verification:
webhook_secret = os.getenv("WASENDER_WEBHOOK_SECRET")
if not api_key:
raise ValueError("WASENDER_API_KEY environment variable not set.")
# Initialize synchronous client using the factory function
sync_client = create_sync_wasender(
api_key=api_key,
personal_access_token=personal_access_token, # Optional, for session management
webhook_secret=webhook_secret, # Optional, for webhook handling
)
# Or initialize directly
# sync_client = WasenderSyncClient(
# api_key=api_key,
# personal_access_token=personal_access_token,
# webhook_secret=webhook_secret,
# )
```
### Asynchronous Client
```python
import os
import asyncio
import httpx # httpx is used by the async client
from wasenderapi import WasenderAsyncClient, create_async_wasender
from wasenderapi.models import RetryConfig
# Required credentials (same as sync client)
api_key = os.getenv("WASENDER_API_KEY")
personal_access_token = os.getenv("WASENDER_PERSONAL_ACCESS_TOKEN")
webhook_secret = os.getenv("WASENDER_WEBHOOK_SECRET")
if not api_key:
raise ValueError("WASENDER_API_KEY environment variable not set.")
async def main():
# Initialize asynchronous client using the factory function
# You can optionally pass your own httpx.AsyncClient instance
# custom_http_client = httpx.AsyncClient()
async_client = create_async_wasender(
api_key=api_key,
personal_access_token=personal_access_token,
webhook_secret=webhook_secret,
)
# Or initialize directly
# async_client = WasenderAsyncClient(
# api_key=api_key,
# personal_access_token=personal_access_token,
# webhook_secret=webhook_secret,
# )
# It's recommended to use the async client as a context manager
# to ensure the underlying httpx.AsyncClient is properly closed.
async with async_client:
# Use the client for API calls
# contacts = await async_client.get_contacts()
# print(contacts)
pass
# If not using 'async with', and if you didn't provide your own httpx_client,
# you might need to manually close the client if it created one internally,
# though the current implementation aims to manage this with __aenter__/__aexit__.
# For safety with direct instantiation without 'async with':
# if async_client._created_http_client and async_client._http_client:
# await async_client._http_client.aclose()
if __name__ == "__main__":
asyncio.run(main())
```
Using the SDK involves calling methods on the initialized client instance. For example, to send a message with the synchronous client:
```python
from wasenderapi.errors import WasenderAPIError
try:
response = sync_client.send_text(to="1234567890", text_body="Hello from Python SDK!")
print(f"Message sent successfully: {response.response.data.message_id}")
if response.rate_limit:
print(f"Rate limit info: Remaining {response.rate_limit.remaining}")
except WasenderAPIError as e:
print(f"API Error: {e.message} (Status: {e.status_code})")
```
For the asynchronous client:
```python
import asyncio
from wasenderapi.errors import WasenderAPIError
# Assuming async_client is initialized within an async context as shown above
async def send_async_text_message(client):
try:
response = await client.send_text(to="1234567890", text_body="Hello from Async Python SDK!")
print(f"Message sent successfully: {response.response.data.message_id}")
if response.rate_limit:
print(f"Rate limit info: Remaining {response.rate_limit.remaining}")
except WasenderAPIError as e:
print(f"API Error: {e.message} (Status: {e.status_code})")
# async def main_async_send_example(): # Renamed to avoid conflict with earlier main
# # ... (async_client initialization as shown in SDK Initialization section)
# # Example:
# # api_key = os.getenv("WASENDER_API_KEY")
# # async_client = create_async_wasender(api_key=api_key)
# async with async_client: # Ensure client is available in this scope
# await send_async_text_message(async_client)
# if __name__ == "__main__":
# asyncio.run(main_async_send_example()) # Example of how to run it
```
## Usage Overview
Using the SDK involves calling methods on the initialized client instance. For example, to send a message with the synchronous client:
```python
from wasenderapi.errors import WasenderAPIError
try:
response = sync_client.send_text(to="1234567890", text_body="Hello from Python SDK!")
print(f"Message sent successfully: {response.response.data.message_id}")
if response.rate_limit:
print(f"Rate limit info: Remaining {response.rate_limit.remaining}")
except WasenderAPIError as e:
print(f"API Error: {e.message} (Status: {e.status_code})")
```
## Authentication
The SDK supports two types of authentication tokens passed during client initialization (e.g., to `create_sync_wasender` or `create_async_wasender`):
1. **API Key (`api_key`)** (Required for most API calls)
- Used for most endpoints, typically those interacting with a specific WhatsApp session (e.g., sending messages, getting contacts for that session).
- This is the primary token for session-specific operations.
2. **Personal Access Token (`personal_access_token`)** (Optional; for account-scoped endpoints)
- Used for account management endpoints that are not tied to a single session, such as listing all WhatsApp sessions under your account, creating new sessions, etc.
- If you only work with a single session and its API key, you might not need this.
The client prioritizes tokens passed directly to its constructor/factory function.
## Core Concepts
### Message Sending: Helpers vs. Generic `send()`
The SDK offers two main ways to send messages:
1. **Specific Helper Methods (Recommended for most common cases):**
Methods like `client.send_text()`, `client.send_image()`, `client.send_video()`, etc., provide a straightforward way to send common message types.
- They accept direct parameters relevant to the message type (e.g., `to: str`, `text_body: str` for `send_text`; `to: str`, `url: str`, `caption: Optional[str]` for `send_image`).
- These methods internally construct the necessary payload structure and Pydantic models.
- They simplify the sending process as you don't need to manually create payload objects.
- Example: `sync_client.send_text(to="PHONE_NUMBER", text_body="Hello!")`
- Example: `await async_client.send_image(to="PHONE_NUMBER", url="http://example.com/image.jpg", caption="My Image")`
2. **Generic `client.send(payload: BaseMessage)` Method:**
- This method provides maximum flexibility and is used for:
- Sending message types that might not have dedicated helper methods.
- Scenarios where you prefer to construct the message payload object (an instance of a class derived from `BaseMessage` in `./models.py`) yourself.
- It requires you to import the appropriate Pydantic model for your message (e.g., `TextOnlyMessage`, `ImageMessage` from `wasenderapi.models`), instantiate it with the necessary data, and then pass it to `client.send()`.
- Example:
```python
from wasenderapi.models import TextOnlyMessage # Or ImageMessage, etc.
# For sync client:
# msg_payload = TextOnlyMessage(to="PHONE_NUMBER", text={"body": "Hello via generic send"})
# response = sync_client.send(msg_payload)
# For async client:
# msg_payload = TextOnlyMessage(to="PHONE_NUMBER", text={"body": "Hello via generic send"})
# response = await client.send(msg_payload)
```
While Pydantic models are used internally by the helper methods and are essential for the generic `send()` method and webhook event parsing, direct interaction with these models for sending common messages is now optional thanks to the simplified helper methods.
### Error Handling
API errors are raised as instances of `WasenderAPIError` (from `wasenderapi.errors`). This exception object includes attributes such as:
- `status_code` (int): The HTTP status code of the error response.
- `api_message` (Optional[str]): The error message from the Wasender API.
- `error_details` (Optional[WasenderErrorDetail]): Further details about the error, if provided by the API.
- `rate_limit` (Optional[RateLimitInfo]): Rate limit information at the time of the error.
### Rate Limiting
Successful API response objects (e.g., `WasenderSendResult`, `WasenderSession`) and `WasenderAPIError` instances may include a `rate_limit` attribute. This attribute is an instance of the `RateLimitInfo` Pydantic model (from `wasenderapi.models`) and provides details about your current API usage limits: `limit`, `remaining`, and `reset_timestamp`.
Rate limit information is primarily expected for `/send-message` related calls but might be present or `None` for other endpoints.
### Webhooks
The Python SDK provides `client.handle_webhook_event(headers: dict, raw_body: bytes, webhook_secret: Optional[str] = None)` for processing incoming webhooks. This method:
1. Verifies the webhook signature using the provided `X-Wasender-Signature` header and the `webhook_secret`.
- The `webhook_secret` can be passed directly to this method or pre-configured on the `WasenderClient` instance during initialization.
2. Parses the validated `raw_body` (which should be the raw bytes of the request body) into a Pydantic `WasenderWebhookEvent` model (a `Union` of specific event types like `MessagesUpsertData`, `SessionStatusData`, etc., defined in `wasenderapi.webhook`).
Unlike some other SDKs, you don't need to implement a separate request adapter; simply pass the necessary request components (headers dictionary and raw body bytes) to the method.
## Usage Examples
This SDK provides a comprehensive suite of functionalities. Below is an overview with links to detailed documentation for each module. For more comprehensive information on all features, please refer to the files in the [`docs`](./docs/) directory.
### 1. Sending Messages
Send various types of messages including text, media (images, videos, documents, audio, stickers), contact cards, and location pins. The easiest way is to use the specific helper methods.
- **Detailed Documentation & Examples:** [`docs/messages.md`](./docs/messages.md)
#### Using the Synchronous Client (`WasenderSyncClient`)
```python
import os
from wasenderapi import create_sync_wasender # Or WasenderSyncClient directly
from wasenderapi.errors import WasenderAPIError
# Initialize client (ensure WASENDER_API_KEY is set)
api_key = os.getenv("WASENDER_API_KEY", "YOUR_SYNC_API_KEY")
sync_client = create_sync_wasender(api_key=api_key)
def send_sync_messages_example():
if sync_client.api_key == "YOUR_SYNC_API_KEY":
print("Please set your WASENDER_API_KEY to run sync examples.")
return
# Example 1: Sending a Text Message
try:
print("Sending text message (sync)...")
response = sync_client.send_text(
to="YOUR_RECIPIENT_PHONE",
text_body="Hello from Wasender Python SDK (Sync)!"
)
print(f" Text message sent: ID {response.response.data.message_id}, Status: {response.response.message}")
if response.rate_limit: print(f" Rate limit: {response.rate_limit.remaining}")
except WasenderAPIError as e:
print(f" Error sending text: {e.message}")
# Example 2: Sending an Image Message
try:
print("Sending image message (sync)...")
response = sync_client.send_image(
to="YOUR_RECIPIENT_PHONE",
url="https://picsum.photos/seed/wasenderpy_sync/300/200", # Replace with your image URL
caption="Test Image from Sync SDK"
)
print(f" Image message sent: ID {response.response.data.message_id}, Status: {response.response.message}")
if response.rate_limit: print(f" Rate limit: {response.rate_limit.remaining}")
except WasenderAPIError as e:
print(f" Error sending image: {e.message}")
# Add more examples for send_video, send_document, etc. as needed.
# To run this specific example:
# if __name__ == "__main__":
# send_sync_messages_example()
```
#### Using the Asynchronous Client (`WasenderAsyncClient`)
```python
import asyncio
import os
from wasenderapi import create_async_wasender # Or WasenderAsyncClient directly
from wasenderapi.errors import WasenderAPIError
# from wasenderapi.models import TextOnlyMessage # Keep for generic send example if shown
# Initialize client (ensure WASENDER_API_KEY is set)
api_key = os.getenv("WASENDER_API_KEY", "YOUR_ASYNC_API_KEY")
async def send_async_messages_example(async_client):
if async_client.api_key == "YOUR_ASYNC_API_KEY":
print("Please set your WASENDER_API_KEY to run async examples.")
return
# Example 1: Sending a Text Message using helper
try:
print("Sending text message (async)...")
response = await async_client.send_text(
to="YOUR_RECIPIENT_PHONE",
text_body="Hello from Wasender Python SDK (Async)!"
)
print(f" Text message sent: ID {response.response.data.message_id}, Status: {response.response.message}")
if response.rate_limit: print(f" Rate limit: {response.rate_limit.remaining}")
except WasenderAPIError as e:
print(f" Error sending text: {e.message}")
# Example 2: Sending an Image Message using helper
try:
print("Sending image message (async)...")
response = await async_client.send_image(
to="YOUR_RECIPIENT_PHONE",
url="https://picsum.photos/seed/wasenderpy_async/300/200", # Replace with your image URL
caption="Test Image from Async SDK"
)
print(f" Image message sent: ID {response.response.data.message_id}, Status: {response.response.message}")
if response.rate_limit: print(f" Rate limit: {response.rate_limit.remaining}")
except WasenderAPIError as e:
print(f" Error sending image: {e.message}")
# Example 3: Sending a Text Message using generic client.send() (for comparison or advanced use)
# from wasenderapi.models import TextOnlyMessage # Ensure import if using this
# try:
# print("Sending text message via generic send (async)...")
# text_payload = TextOnlyMessage(
# to="YOUR_RECIPIENT_PHONE",
# text={"body": "Hello again via generic send (Async)!"}
# )
# response = await async_client.send(text_payload)
# print(f" Generic text message sent: Status {response.response.message}")
# except WasenderAPIError as e:
# print(f" Error sending generic text: {e.message}")
async def main_run_async_examples():
# It's recommended to use the async client as a context manager.
async_client = create_async_wasender(api_key=api_key)
async with async_client:
await send_async_messages_example(async_client)
# To run this specific example:
# if __name__ == "__main__":
# if api_key == "YOUR_ASYNC_API_KEY":
# print("Please set your WASENDER_API_KEY environment variable to run this example.")
# else:
# asyncio.run(main_run_async_examples())
```
### 2. Managing Contacts
Retrieve your contact list, fetch information about specific contacts, get their profile pictures, and block or unblock contacts.
- **Detailed Documentation & Examples:** [`docs/contacts.md`](./docs/contacts.md)
```python
import asyncio
import os
from wasenderapi import create_async_wasender # Use the factory for async client
from wasenderapi.errors import WasenderAPIError
# api_key = os.getenv("WASENDER_API_KEY", "YOUR_API_KEY_HERE") # Define API key
async def manage_contacts_example():
# Initialize client within the async function or pass it as an argument
# This example assumes api_key is defined in the scope
local_api_key = os.getenv("WASENDER_API_KEY", "YOUR_CONTACTS_API_KEY")
if local_api_key == "YOUR_CONTACTS_API_KEY":
print("Error: WASENDER_API_KEY not set for contacts example.")
return
async with create_async_wasender(api_key=local_api_key) as async_client:
print("\nAttempting to fetch contacts...")
try:
result = await async_client.get_contacts()
if result.response and result.response.data is not None:
contacts = result.response.data
print(f"Successfully fetched {len(contacts)} contacts.")
if contacts:
first_contact = contacts[0]
print(f" First contact - JID: {first_contact.jid}, Name: {first_contact.name or 'N/A'}")
else:
print("No contact data received.")
if result.rate_limit: print(f"Rate limit: {result.rate_limit.remaining}/{result.rate_limit.limit}")
except WasenderAPIError as e:
print(f"API Error fetching contacts: {e.message}")
except Exception as e:
print(f"An unexpected error: {e}")
# To run this example (ensure WASENDER_API_KEY is set):
# if __name__ == "__main__":
# asyncio.run(manage_contacts_example())
```
### 3. Managing Groups
List groups your account is part of, get group metadata, manage participants, and update group settings.
- **Detailed Documentation & Examples:** [`docs/groups.md`](./docs/groups.md)
```python
import asyncio
import os
from wasenderapi import create_async_wasender # Use the factory for async client
from wasenderapi.errors import WasenderAPIError
async def manage_groups_example():
local_api_key = os.getenv("WASENDER_API_KEY", "YOUR_GROUPS_API_KEY")
if local_api_key == "YOUR_GROUPS_API_KEY":
print("Error: WASENDER_API_KEY not set for groups example.")
return
async with create_async_wasender(api_key=local_api_key) as async_client:
print("\nAttempting to fetch groups...")
try:
groups_result = await async_client.get_groups()
if groups_result.response and groups_result.response.data is not None:
groups = groups_result.response.data
print(f"Successfully fetched {len(groups)} groups.")
if groups:
first_group = groups[0]
print(f" First group - JID: {first_group.jid}, Subject: {first_group.subject}")
# Example: Get metadata for the first group
print(f" Fetching metadata for group: {first_group.jid}...")
try:
metadata_result = await async_client.get_group_metadata(group_jid=first_group.jid)
if metadata_result.response and metadata_result.response.data:
metadata = metadata_result.response.data
participant_count = len(metadata.participants) if metadata.participants else 0
print(f" Group Subject: {metadata.subject}, Participants: {participant_count}")
else:
print(f" Could not retrieve metadata for group {first_group.jid}.")
if metadata_result.rate_limit: print(f" Metadata Rate limit: {metadata_result.rate_limit.remaining}")
except WasenderAPIError as e_meta:
print(f" API Error fetching group metadata: {e_meta.message}")
else:
print("No group data received.")
if groups_result.rate_limit: print(f"Groups List Rate limit: {groups_result.rate_limit.remaining}")
except WasenderAPIError as e:
print(f"API Error fetching groups list: {e.message}")
except Exception as e:
print(f"An unexpected error: {e}")
# To run this example (ensure WASENDER_API_KEY is set):
# if __name__ == "__main__":
# asyncio.run(manage_groups_example())
```
### 4. Sending Messages to WhatsApp Channels
Send text messages to WhatsApp Channels.
- **Detailed Documentation & Examples:** [`docs/channel.md`](./docs/channel.md)
```python
import asyncio
import os
from wasenderapi import create_async_wasender # Use factory for async client
from wasenderapi.errors import WasenderAPIError
from wasenderapi.models import ChannelTextMessage # This model is for generic send()
async def send_to_channel_example(channel_jid: str, text_message: str):
local_api_key = os.getenv("WASENDER_API_KEY", "YOUR_CHANNEL_API_KEY")
if local_api_key == "YOUR_CHANNEL_API_KEY":
print("Error: WASENDER_API_KEY not set for channel example.")
return
# For sending to channels, the SDK might have a specific helper or use the generic send.
# This example assumes use of generic send() with ChannelTextMessage model as shown previously.
# If a client.send_channel_text() helper exists, prefer that.
async with create_async_wasender(api_key=local_api_key) as async_client:
print(f"\nAttempting to send to WhatsApp Channel {channel_jid}...")
try:
payload = ChannelTextMessage(
to=channel_jid,
message_type="text", # Assuming model requires this
text=text_message
)
# Using generic send method for channel messages as per original example
result = await async_client.send(payload)
print(f"Message sent to channel successfully.")
if result.response: print(f" Message ID: {result.response.message_id}, Status: {result.response.message}")
if result.rate_limit: print(f"Rate limit: {result.rate_limit.remaining}/{result.rate_limit.limit}")
except WasenderAPIError as e:
print(f"API Error sending to channel: {e.message}")
except Exception as e:
print(f"An unexpected error: {e}")
# To run this example (ensure WASENDER_API_KEY and TEST_CHANNEL_JID are set):
# async def main_run_channel_example():
# test_channel_jid = os.getenv("TEST_CHANNEL_JID", "YOUR_CHANNEL_JID@newsletter")
# message = "Hello Channel from SDK Example!"
# if test_channel_jid == "YOUR_CHANNEL_JID@newsletter":
# print("Please set a valid TEST_CHANNEL_JID environment variable.")
# return
# await send_to_channel_example(channel_jid=test_channel_jid, text_message=message)
# if __name__ == "__main__":
# asyncio.run(main_run_channel_example())
```
### 5. Handling Incoming Webhooks
Process real-time events from Wasender. The `handle_webhook_event` method is available on both sync and async clients, but it is an `async` method in both cases.
- **Detailed Documentation & Examples:** [`docs/webhook.md`](./docs/webhook.md)
```python
import os
import json # For pretty printing
# Use the appropriate client factory
from wasenderapi import create_sync_wasender, create_async_wasender
from wasenderapi.webhook import WasenderWebhookEvent, WasenderWebhookEventType
from wasenderapi.errors import SignatureVerificationError, WasenderAPIError
# Client initialization for webhook handling:
# Webhook secret is the key component here.
webhook_secret_from_env = os.getenv("WASENDER_WEBHOOK_SECRET")
if not webhook_secret_from_env:
print("CRITICAL: WASENDER_WEBHOOK_SECRET environment variable is not set.")
# webhook_secret_from_env = "YOUR_PLACEHOLDER_WEBHOOK_SECRET" # For structure viewing
# You can use either sync or async client to access handle_webhook_event.
# However, since handle_webhook_event itself is async, using it in a purely
# synchronous web framework like standard Flask requires special handling (e.g., asyncio.run_coroutine_threadsafe).
# For an async framework (like FastAPI, Quart), you'd use the async client.
# Example: Initialize a sync client for its config, but note the method is async.
sync_client_for_webhook = create_sync_wasender(
api_key="DUMMY_API_KEY_NOT_USED_FOR_WEBHOOK_LOGIC", # API key not strictly needed for webhook verification logic
webhook_secret=webhook_secret_from_env
)
# --- Conceptual Example: How to use with a web framework (e.g., Flask) ---
# This illustrates the logic. Actual integration needs to handle the async nature
# of handle_webhook_event within the sync framework.
#
# from flask import Flask, request, jsonify
# import asyncio
#
# app = Flask(__name__)
#
# def run_async_in_sync(coro):
# # A simple way to run an async function from sync code (for demonstration)
# # In production, use framework-specific solutions or proper async event loop management.
# loop = asyncio.new_event_loop()
# asyncio.set_event_loop(loop)
# return loop.run_until_complete(coro)
#
# @app.route("/wasender-webhook", methods=["POST"])
# def flask_webhook_handler():
# raw_body_bytes = request.get_data()
# headers_dict = dict(request.headers)
#
# if not sync_client_for_webhook.webhook_secret:
# print("Error: Webhook secret not configured.")
# return jsonify({"error": "Webhook secret not configured"}), 500
#
# try:
# # handle_webhook_event is async, so needs to be run in an event loop
# event: WasenderWebhookEvent = run_async_in_sync(
# sync_client_for_webhook.handle_webhook_event(
# request_body_bytes=raw_body_bytes, # Pass raw_body_bytes here
# signature_header=headers_dict.get(WasenderWebhookEvent.SIGNATURE_HEADER) # Pass correct header
# )
# )
# print(f"Webhook Type: {event.event_type.value}")
# # ... (process event.data based on event.event_type) ...
# return jsonify({"status": "success"}), 200
# except SignatureVerificationError as e:
# return jsonify({"error": "Signature verification failed"}), 400
# except WasenderAPIError as e: # Handles other SDK errors like bad payload
# return jsonify({"error": f"Webhook processing error: {e.message}"}), 400
# except Exception as e:
# return jsonify({"error": "Internal server error"}), 500
# Dummy data for local testing (if __name__ == "__main__")
# ... (The existing dummy data and call simulation can be kept but adapted ...)
# ... ensure to call sync_client_for_webhook.handle_webhook_event ...
# if __name__ == "__main__":
# # ... (simulation code using sync_client_for_webhook.handle_webhook_event)
# # Remember the async nature when simulating the call directly.
# pass
```
### 6. Managing WhatsApp Sessions
Create, list, update, delete sessions, connect/disconnect, get QR codes, and check session status. Listing all sessions or creating new ones typically requires a `personal_access_token`.
- **Detailed Documentation & Examples:** [`docs/sessions.md`](./docs/sessions.md)
```python
import asyncio
import os
from wasenderapi import create_async_wasender # Use factory for async client
from wasenderapi.errors import WasenderAPIError
async def manage_whatsapp_sessions_example():
local_api_key = os.getenv("WASENDER_API_KEY", "YOUR_SESSIONS_API_KEY")
# For listing all sessions or creating new ones, a personal access token is often needed.
local_personal_access_token = os.getenv("WASENDER_PERSONAL_ACCESS_TOKEN")
if local_api_key == "YOUR_SESSIONS_API_KEY":
print("Error: WASENDER_API_KEY not set for sessions example.")
return
# Client for session-specific actions (uses api_key of that session)
# Client for account-level session actions (uses personal_access_token)
# We'll use one client, assuming personal_access_token is for listing/creating if needed,
# and api_key for specific session operations if that's how the SDK/API is structured.
# The create_async_wasender can take both.
async with create_async_wasender(api_key=local_api_key, personal_access_token=local_personal_access_token) as async_client:
if not local_personal_access_token:
print("Warning: WASENDER_PERSONAL_ACCESS_TOKEN not set. Listing all sessions might fail or be restricted.")
print("\nAttempting to list WhatsApp sessions...")
try:
# get_all_whatsapp_sessions is the method for listing all sessions
list_sessions_result = await async_client.get_all_whatsapp_sessions()
if list_sessions_result.response and list_sessions_result.response.data is not None:
sessions = list_sessions_result.response.data
print(f"Successfully fetched {len(sessions)} session(s).")
if sessions:
for session_info in sessions:
print(f" Session ID: {session_info.session_id}, Status: {session_info.status.value if session_info.status else 'N/A'}")
else:
print(" No active sessions found for this account.")
else:
print("No session data received.")
if list_sessions_result.rate_limit: print(f"List Sessions Rate limit: {list_sessions_result.rate_limit.remaining}")
# Example for get_session_status (uses api_key of a specific session)
# if sessions:
# target_session_id = sessions[0].session_id # This is numeric ID from the list
# # To get status, you typically use the API key associated with *that* session_id,
# # which might be different from the local_api_key if it's a PAT.
# # For this example, we assume async_client was initialized with the correct session's API key
# # if we were to uncomment and run the following. Or, one might re-initialize a client
# # specifically for that session if its API key is known.
# print(f"\nAttempting to get status for session: {target_session_id}")
# try:
# # Assuming local_api_key IS the key for this specific session if we call get_session_status
# # If not, this call might not be correctly authorized.
# status_result = await async_client.get_session_status(session_id=str(target_session_id)) # Ensure session_id is string if required
# if status_result.response and status_result.response.data:
# print(f" Status for {target_session_id}: {status_result.response.data.status.value}")
# except WasenderAPIError as e_status:
# print(f" Error getting status for {target_session_id}: {e_status.message}")
except WasenderAPIError as e:
print(f"API Error managing sessions: {e.message}")
except Exception as e:
print(f"An unexpected error: {e}")
# To run this example (ensure WASENDER_API_KEY and optionally WASENDER_PERSONAL_ACCESS_TOKEN are set):
# if __name__ == "__main__":
# asyncio.run(manage_whatsapp_sessions_example())
```
## Advanced Topics
### Configuring Retries
The SDK supports automatic retries, primarily for handling HTTP 429 (Too Many Requests) errors. This is configured via the `RetryConfig` object passed to `retry_options` during client initialization.
```python
from wasenderapi import create_sync_wasender, create_async_wasender
from wasenderapi.models import RetryConfig
# Configure retries: enable and set max retries
retry_settings = RetryConfig(enabled=True, max_retries=3)
# For synchronous client
sync_client_with_retries = create_sync_wasender(
api_key="YOUR_API_KEY",
retry_options=retry_settings
)
# For asynchronous client
async_client_with_retries = create_async_wasender(
api_key="YOUR_API_KEY",
retry_options=retry_settings
)
# Example usage (sync)
# try:
# sync_client_with_retries.send_text(to="PHONE", text_body="Test with retries")
# except WasenderAPIError as e:
# print(f"Failed after retries: {e}")
```
By default, retries are disabled.
## Contributing
Contributions are welcome! Please feel free to submit issues, fork the repository, and create pull requests.
## License
This SDK is released under the [MIT License](./LICENSE).
Raw data
{
"_id": null,
"home_page": null,
"name": "wasenderapi",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "api, chatbot, messaging, sdk, wasender, whatsapp",
"author": null,
"author_email": "YonkoSam <absamlani@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/a2/76/a75eeecbd1283b2ef0a6e52844012bfc498f20723af838252f44a710f3ba/wasenderapi-0.3.4.tar.gz",
"platform": null,
"description": "# Wasender API Python SDK\n\n**Python SDK Author:** YonkoSam\n**Original Node.js SDK Author:** Shreshth Arora\n\n[](https://pypi.org/project/wasenderapi/)\n[](https://pypi.org/project/wasenderapi/)\n[](LICENSE)\n[](https://www.python.org/)\n[](https://github.com/YonkoSam/wasenderapi-python/actions/workflows/ci.yml)\n\nA lightweight and robust Python SDK for interacting with the Wasender API ([https://www.wasenderapi.com](https://www.wasenderapi.com)). This SDK simplifies sending various types of WhatsApp messages, managing contacts and groups, handling session statuses, and processing incoming webhooks.\n\n## Features\n\n- **Pydantic Models:** Leverages Pydantic for robust request/response validation and serialization, especially for the generic `send()` method and webhook event parsing.\n- **Message Sending:**\n - Simplified helper methods (e.g., `client.send_text(to=\"...\", text_body=\"...\")`, `client.send_image(to=\"...\", url=\"...\", caption=\"...\")`) that accept direct parameters for common message types.\n - Generic `client.send(payload: BaseMessage)` method for advanced use cases or less common message types, accepting a Pydantic model.\n - Support for text, image, video, document, audio, sticker, contact card, and location messages.\n- **Contact Management:** List, retrieve details, get profile pictures, block, and unblock contacts.\n- **Group Management:** List groups, fetch metadata, manage participants (add/remove), and update group settings.\n- **Channel Messaging:** Send text messages to WhatsApp Channels.\n- **Session Management:** Create, list, update, delete sessions, connect/disconnect, get QR codes, and check session status.\n- **Webhook Handling:** Securely verify and parse incoming webhook events from Wasender using Pydantic models.\n- **Error Handling:** Comprehensive `WasenderAPIError` class with detailed error information.\n- **Rate Limiting:** Access to rate limit information on API responses.\n- **Retry Mechanism:** Optional automatic retries for rate-limited requests (HTTP 429) via `RetryConfig`.\n- **Customizable HTTP Client:** Allows providing a custom `httpx.AsyncClient` instance for the asynchronous client.\n\n## Prerequisites\n\n- Python (version 3.8 or higher recommended).\n- A Wasender API Key from [https://www.wasenderapi.com](https://www.wasenderapi.com).\n- If using webhooks:\n - A publicly accessible HTTPS URL for your webhook endpoint.\n - A Webhook Secret generated from the Wasender dashboard.\n\n## Installation\n\n```bash\npip install wasenderapi\n```\n\n## SDK Initialization\n\nThe SDK now provides both a synchronous and an asynchronous client.\n\n### Synchronous Client\n\n```python\nimport os\nfrom wasenderapi import WasenderSyncClient, create_sync_wasender\nfrom wasenderapi.models import RetryConfig\n\n# Required credentials\napi_key = os.getenv(\"WASENDER_API_KEY\")\n# For account-scoped endpoints like session management:\npersonal_access_token = os.getenv(\"WASENDER_PERSONAL_ACCESS_TOKEN\") \n# For webhook verification:\nwebhook_secret = os.getenv(\"WASENDER_WEBHOOK_SECRET\") \n\nif not api_key:\n raise ValueError(\"WASENDER_API_KEY environment variable not set.\")\n\n# Initialize synchronous client using the factory function\nsync_client = create_sync_wasender(\n api_key=api_key,\n personal_access_token=personal_access_token, # Optional, for session management\n webhook_secret=webhook_secret, # Optional, for webhook handling\n)\n\n# Or initialize directly\n# sync_client = WasenderSyncClient(\n# api_key=api_key,\n# personal_access_token=personal_access_token,\n# webhook_secret=webhook_secret,\n# )\n```\n\n### Asynchronous Client\n\n```python\nimport os\nimport asyncio\nimport httpx # httpx is used by the async client\nfrom wasenderapi import WasenderAsyncClient, create_async_wasender\nfrom wasenderapi.models import RetryConfig\n\n# Required credentials (same as sync client)\napi_key = os.getenv(\"WASENDER_API_KEY\")\npersonal_access_token = os.getenv(\"WASENDER_PERSONAL_ACCESS_TOKEN\")\nwebhook_secret = os.getenv(\"WASENDER_WEBHOOK_SECRET\")\n\nif not api_key:\n raise ValueError(\"WASENDER_API_KEY environment variable not set.\")\n\nasync def main():\n # Initialize asynchronous client using the factory function\n # You can optionally pass your own httpx.AsyncClient instance\n # custom_http_client = httpx.AsyncClient()\n async_client = create_async_wasender(\n api_key=api_key,\n personal_access_token=personal_access_token,\n webhook_secret=webhook_secret,\n )\n\n # Or initialize directly\n # async_client = WasenderAsyncClient(\n # api_key=api_key,\n # personal_access_token=personal_access_token,\n # webhook_secret=webhook_secret,\n # )\n\n # It's recommended to use the async client as a context manager\n # to ensure the underlying httpx.AsyncClient is properly closed.\n async with async_client:\n # Use the client for API calls\n # contacts = await async_client.get_contacts()\n # print(contacts)\n pass\n\n # If not using 'async with', and if you didn't provide your own httpx_client,\n # you might need to manually close the client if it created one internally,\n # though the current implementation aims to manage this with __aenter__/__aexit__.\n # For safety with direct instantiation without 'async with':\n # if async_client._created_http_client and async_client._http_client:\n # await async_client._http_client.aclose()\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\nUsing the SDK involves calling methods on the initialized client instance. For example, to send a message with the synchronous client:\n\n```python\nfrom wasenderapi.errors import WasenderAPIError\n\ntry:\n response = sync_client.send_text(to=\"1234567890\", text_body=\"Hello from Python SDK!\")\n print(f\"Message sent successfully: {response.response.data.message_id}\")\n if response.rate_limit:\n print(f\"Rate limit info: Remaining {response.rate_limit.remaining}\")\nexcept WasenderAPIError as e:\n print(f\"API Error: {e.message} (Status: {e.status_code})\")\n```\n\nFor the asynchronous client:\n\n```python\nimport asyncio\nfrom wasenderapi.errors import WasenderAPIError\n\n# Assuming async_client is initialized within an async context as shown above\nasync def send_async_text_message(client):\n try:\n response = await client.send_text(to=\"1234567890\", text_body=\"Hello from Async Python SDK!\")\n print(f\"Message sent successfully: {response.response.data.message_id}\")\n if response.rate_limit:\n print(f\"Rate limit info: Remaining {response.rate_limit.remaining}\")\n except WasenderAPIError as e:\n print(f\"API Error: {e.message} (Status: {e.status_code})\")\n\n# async def main_async_send_example(): # Renamed to avoid conflict with earlier main\n# # ... (async_client initialization as shown in SDK Initialization section)\n# # Example: \n# # api_key = os.getenv(\"WASENDER_API_KEY\")\n# # async_client = create_async_wasender(api_key=api_key)\n# async with async_client: # Ensure client is available in this scope\n# await send_async_text_message(async_client)\n\n# if __name__ == \"__main__\":\n# asyncio.run(main_async_send_example()) # Example of how to run it\n```\n\n## Usage Overview\n\nUsing the SDK involves calling methods on the initialized client instance. For example, to send a message with the synchronous client:\n\n```python\nfrom wasenderapi.errors import WasenderAPIError\n\ntry:\n response = sync_client.send_text(to=\"1234567890\", text_body=\"Hello from Python SDK!\")\n print(f\"Message sent successfully: {response.response.data.message_id}\")\n if response.rate_limit:\n print(f\"Rate limit info: Remaining {response.rate_limit.remaining}\")\nexcept WasenderAPIError as e:\n print(f\"API Error: {e.message} (Status: {e.status_code})\")\n```\n\n## Authentication\n\nThe SDK supports two types of authentication tokens passed during client initialization (e.g., to `create_sync_wasender` or `create_async_wasender`):\n\n1. **API Key (`api_key`)** (Required for most API calls)\n - Used for most endpoints, typically those interacting with a specific WhatsApp session (e.g., sending messages, getting contacts for that session).\n - This is the primary token for session-specific operations.\n\n2. **Personal Access Token (`personal_access_token`)** (Optional; for account-scoped endpoints)\n - Used for account management endpoints that are not tied to a single session, such as listing all WhatsApp sessions under your account, creating new sessions, etc.\n - If you only work with a single session and its API key, you might not need this.\n\nThe client prioritizes tokens passed directly to its constructor/factory function.\n\n## Core Concepts\n\n### Message Sending: Helpers vs. Generic `send()`\n\nThe SDK offers two main ways to send messages:\n\n1. **Specific Helper Methods (Recommended for most common cases):**\n Methods like `client.send_text()`, `client.send_image()`, `client.send_video()`, etc., provide a straightforward way to send common message types.\n - They accept direct parameters relevant to the message type (e.g., `to: str`, `text_body: str` for `send_text`; `to: str`, `url: str`, `caption: Optional[str]` for `send_image`).\n - These methods internally construct the necessary payload structure and Pydantic models.\n - They simplify the sending process as you don't need to manually create payload objects.\n - Example: `sync_client.send_text(to=\"PHONE_NUMBER\", text_body=\"Hello!\")`\n - Example: `await async_client.send_image(to=\"PHONE_NUMBER\", url=\"http://example.com/image.jpg\", caption=\"My Image\")`\n\n2. **Generic `client.send(payload: BaseMessage)` Method:**\n - This method provides maximum flexibility and is used for:\n - Sending message types that might not have dedicated helper methods.\n - Scenarios where you prefer to construct the message payload object (an instance of a class derived from `BaseMessage` in `./models.py`) yourself.\n - It requires you to import the appropriate Pydantic model for your message (e.g., `TextOnlyMessage`, `ImageMessage` from `wasenderapi.models`), instantiate it with the necessary data, and then pass it to `client.send()`.\n - Example:\n ```python\n from wasenderapi.models import TextOnlyMessage # Or ImageMessage, etc.\n # For sync client:\n # msg_payload = TextOnlyMessage(to=\"PHONE_NUMBER\", text={\"body\": \"Hello via generic send\"})\n # response = sync_client.send(msg_payload)\n # For async client:\n # msg_payload = TextOnlyMessage(to=\"PHONE_NUMBER\", text={\"body\": \"Hello via generic send\"})\n # response = await client.send(msg_payload)\n ```\n\nWhile Pydantic models are used internally by the helper methods and are essential for the generic `send()` method and webhook event parsing, direct interaction with these models for sending common messages is now optional thanks to the simplified helper methods.\n\n### Error Handling\n\nAPI errors are raised as instances of `WasenderAPIError` (from `wasenderapi.errors`). This exception object includes attributes such as:\n- `status_code` (int): The HTTP status code of the error response.\n- `api_message` (Optional[str]): The error message from the Wasender API.\n- `error_details` (Optional[WasenderErrorDetail]): Further details about the error, if provided by the API.\n- `rate_limit` (Optional[RateLimitInfo]): Rate limit information at the time of the error.\n\n### Rate Limiting\n\nSuccessful API response objects (e.g., `WasenderSendResult`, `WasenderSession`) and `WasenderAPIError` instances may include a `rate_limit` attribute. This attribute is an instance of the `RateLimitInfo` Pydantic model (from `wasenderapi.models`) and provides details about your current API usage limits: `limit`, `remaining`, and `reset_timestamp`.\n\nRate limit information is primarily expected for `/send-message` related calls but might be present or `None` for other endpoints.\n\n### Webhooks\n\nThe Python SDK provides `client.handle_webhook_event(headers: dict, raw_body: bytes, webhook_secret: Optional[str] = None)` for processing incoming webhooks. This method:\n1. Verifies the webhook signature using the provided `X-Wasender-Signature` header and the `webhook_secret`.\n - The `webhook_secret` can be passed directly to this method or pre-configured on the `WasenderClient` instance during initialization.\n2. Parses the validated `raw_body` (which should be the raw bytes of the request body) into a Pydantic `WasenderWebhookEvent` model (a `Union` of specific event types like `MessagesUpsertData`, `SessionStatusData`, etc., defined in `wasenderapi.webhook`).\n\nUnlike some other SDKs, you don't need to implement a separate request adapter; simply pass the necessary request components (headers dictionary and raw body bytes) to the method.\n\n## Usage Examples\n\nThis SDK provides a comprehensive suite of functionalities. Below is an overview with links to detailed documentation for each module. For more comprehensive information on all features, please refer to the files in the [`docs`](./docs/) directory.\n\n### 1. Sending Messages\n\nSend various types of messages including text, media (images, videos, documents, audio, stickers), contact cards, and location pins. The easiest way is to use the specific helper methods.\n\n- **Detailed Documentation & Examples:** [`docs/messages.md`](./docs/messages.md)\n\n#### Using the Synchronous Client (`WasenderSyncClient`)\n\n```python\nimport os\nfrom wasenderapi import create_sync_wasender # Or WasenderSyncClient directly\nfrom wasenderapi.errors import WasenderAPIError\n\n# Initialize client (ensure WASENDER_API_KEY is set)\napi_key = os.getenv(\"WASENDER_API_KEY\", \"YOUR_SYNC_API_KEY\")\nsync_client = create_sync_wasender(api_key=api_key)\n\ndef send_sync_messages_example():\n if sync_client.api_key == \"YOUR_SYNC_API_KEY\":\n print(\"Please set your WASENDER_API_KEY to run sync examples.\")\n return\n\n # Example 1: Sending a Text Message\n try:\n print(\"Sending text message (sync)...\")\n response = sync_client.send_text(\n to=\"YOUR_RECIPIENT_PHONE\", \n text_body=\"Hello from Wasender Python SDK (Sync)!\"\n )\n print(f\" Text message sent: ID {response.response.data.message_id}, Status: {response.response.message}\")\n if response.rate_limit: print(f\" Rate limit: {response.rate_limit.remaining}\")\n except WasenderAPIError as e:\n print(f\" Error sending text: {e.message}\")\n\n # Example 2: Sending an Image Message\n try:\n print(\"Sending image message (sync)...\")\n response = sync_client.send_image(\n to=\"YOUR_RECIPIENT_PHONE\",\n url=\"https://picsum.photos/seed/wasenderpy_sync/300/200\", # Replace with your image URL\n caption=\"Test Image from Sync SDK\"\n )\n print(f\" Image message sent: ID {response.response.data.message_id}, Status: {response.response.message}\")\n if response.rate_limit: print(f\" Rate limit: {response.rate_limit.remaining}\")\n except WasenderAPIError as e:\n print(f\" Error sending image: {e.message}\")\n\n # Add more examples for send_video, send_document, etc. as needed.\n\n# To run this specific example:\n# if __name__ == \"__main__\":\n# send_sync_messages_example()\n```\n\n#### Using the Asynchronous Client (`WasenderAsyncClient`)\n\n```python\nimport asyncio\nimport os\nfrom wasenderapi import create_async_wasender # Or WasenderAsyncClient directly\nfrom wasenderapi.errors import WasenderAPIError\n# from wasenderapi.models import TextOnlyMessage # Keep for generic send example if shown\n\n# Initialize client (ensure WASENDER_API_KEY is set)\napi_key = os.getenv(\"WASENDER_API_KEY\", \"YOUR_ASYNC_API_KEY\")\n\nasync def send_async_messages_example(async_client):\n if async_client.api_key == \"YOUR_ASYNC_API_KEY\":\n print(\"Please set your WASENDER_API_KEY to run async examples.\")\n return\n\n # Example 1: Sending a Text Message using helper\n try:\n print(\"Sending text message (async)...\")\n response = await async_client.send_text(\n to=\"YOUR_RECIPIENT_PHONE\", \n text_body=\"Hello from Wasender Python SDK (Async)!\"\n )\n print(f\" Text message sent: ID {response.response.data.message_id}, Status: {response.response.message}\")\n if response.rate_limit: print(f\" Rate limit: {response.rate_limit.remaining}\")\n except WasenderAPIError as e:\n print(f\" Error sending text: {e.message}\")\n\n # Example 2: Sending an Image Message using helper\n try:\n print(\"Sending image message (async)...\")\n response = await async_client.send_image(\n to=\"YOUR_RECIPIENT_PHONE\",\n url=\"https://picsum.photos/seed/wasenderpy_async/300/200\", # Replace with your image URL\n caption=\"Test Image from Async SDK\"\n )\n print(f\" Image message sent: ID {response.response.data.message_id}, Status: {response.response.message}\")\n if response.rate_limit: print(f\" Rate limit: {response.rate_limit.remaining}\")\n except WasenderAPIError as e:\n print(f\" Error sending image: {e.message}\")\n\n # Example 3: Sending a Text Message using generic client.send() (for comparison or advanced use)\n # from wasenderapi.models import TextOnlyMessage # Ensure import if using this\n # try:\n # print(\"Sending text message via generic send (async)...\")\n # text_payload = TextOnlyMessage(\n # to=\"YOUR_RECIPIENT_PHONE\",\n # text={\"body\": \"Hello again via generic send (Async)!\"} \n # )\n # response = await async_client.send(text_payload)\n # print(f\" Generic text message sent: Status {response.response.message}\")\n # except WasenderAPIError as e:\n # print(f\" Error sending generic text: {e.message}\")\n\nasync def main_run_async_examples():\n # It's recommended to use the async client as a context manager.\n async_client = create_async_wasender(api_key=api_key)\n async with async_client:\n await send_async_messages_example(async_client)\n\n# To run this specific example:\n# if __name__ == \"__main__\":\n# if api_key == \"YOUR_ASYNC_API_KEY\":\n# print(\"Please set your WASENDER_API_KEY environment variable to run this example.\")\n# else:\n# asyncio.run(main_run_async_examples())\n```\n\n### 2. Managing Contacts\n\nRetrieve your contact list, fetch information about specific contacts, get their profile pictures, and block or unblock contacts.\n\n- **Detailed Documentation & Examples:** [`docs/contacts.md`](./docs/contacts.md)\n\n```python\nimport asyncio\nimport os\nfrom wasenderapi import create_async_wasender # Use the factory for async client\nfrom wasenderapi.errors import WasenderAPIError\n\n# api_key = os.getenv(\"WASENDER_API_KEY\", \"YOUR_API_KEY_HERE\") # Define API key\n\nasync def manage_contacts_example():\n # Initialize client within the async function or pass it as an argument\n # This example assumes api_key is defined in the scope\n local_api_key = os.getenv(\"WASENDER_API_KEY\", \"YOUR_CONTACTS_API_KEY\")\n if local_api_key == \"YOUR_CONTACTS_API_KEY\":\n print(\"Error: WASENDER_API_KEY not set for contacts example.\")\n return\n\n async with create_async_wasender(api_key=local_api_key) as async_client:\n print(\"\\nAttempting to fetch contacts...\")\n try:\n result = await async_client.get_contacts()\n \n if result.response and result.response.data is not None:\n contacts = result.response.data\n print(f\"Successfully fetched {len(contacts)} contacts.\")\n if contacts:\n first_contact = contacts[0]\n print(f\" First contact - JID: {first_contact.jid}, Name: {first_contact.name or 'N/A'}\")\n else:\n print(\"No contact data received.\")\n\n if result.rate_limit: print(f\"Rate limit: {result.rate_limit.remaining}/{result.rate_limit.limit}\")\n\n except WasenderAPIError as e:\n print(f\"API Error fetching contacts: {e.message}\")\n except Exception as e:\n print(f\"An unexpected error: {e}\")\n\n# To run this example (ensure WASENDER_API_KEY is set):\n# if __name__ == \"__main__\":\n# asyncio.run(manage_contacts_example())\n```\n\n### 3. Managing Groups\n\nList groups your account is part of, get group metadata, manage participants, and update group settings.\n\n- **Detailed Documentation & Examples:** [`docs/groups.md`](./docs/groups.md)\n\n```python\nimport asyncio\nimport os\nfrom wasenderapi import create_async_wasender # Use the factory for async client\nfrom wasenderapi.errors import WasenderAPIError\n\nasync def manage_groups_example():\n local_api_key = os.getenv(\"WASENDER_API_KEY\", \"YOUR_GROUPS_API_KEY\")\n if local_api_key == \"YOUR_GROUPS_API_KEY\":\n print(\"Error: WASENDER_API_KEY not set for groups example.\")\n return\n\n async with create_async_wasender(api_key=local_api_key) as async_client:\n print(\"\\nAttempting to fetch groups...\")\n try:\n groups_result = await async_client.get_groups()\n \n if groups_result.response and groups_result.response.data is not None:\n groups = groups_result.response.data\n print(f\"Successfully fetched {len(groups)} groups.\")\n \n if groups:\n first_group = groups[0]\n print(f\" First group - JID: {first_group.jid}, Subject: {first_group.subject}\")\n\n # Example: Get metadata for the first group\n print(f\" Fetching metadata for group: {first_group.jid}...\")\n try:\n metadata_result = await async_client.get_group_metadata(group_jid=first_group.jid)\n if metadata_result.response and metadata_result.response.data:\n metadata = metadata_result.response.data\n participant_count = len(metadata.participants) if metadata.participants else 0\n print(f\" Group Subject: {metadata.subject}, Participants: {participant_count}\")\n else:\n print(f\" Could not retrieve metadata for group {first_group.jid}.\")\n if metadata_result.rate_limit: print(f\" Metadata Rate limit: {metadata_result.rate_limit.remaining}\")\n except WasenderAPIError as e_meta:\n print(f\" API Error fetching group metadata: {e_meta.message}\")\n else:\n print(\"No group data received.\")\n\n if groups_result.rate_limit: print(f\"Groups List Rate limit: {groups_result.rate_limit.remaining}\")\n\n except WasenderAPIError as e:\n print(f\"API Error fetching groups list: {e.message}\")\n except Exception as e:\n print(f\"An unexpected error: {e}\")\n\n# To run this example (ensure WASENDER_API_KEY is set):\n# if __name__ == \"__main__\":\n# asyncio.run(manage_groups_example())\n```\n\n### 4. Sending Messages to WhatsApp Channels\n\nSend text messages to WhatsApp Channels.\n\n- **Detailed Documentation & Examples:** [`docs/channel.md`](./docs/channel.md)\n\n```python\nimport asyncio\nimport os\nfrom wasenderapi import create_async_wasender # Use factory for async client\nfrom wasenderapi.errors import WasenderAPIError\nfrom wasenderapi.models import ChannelTextMessage # This model is for generic send()\n\nasync def send_to_channel_example(channel_jid: str, text_message: str):\n local_api_key = os.getenv(\"WASENDER_API_KEY\", \"YOUR_CHANNEL_API_KEY\")\n if local_api_key == \"YOUR_CHANNEL_API_KEY\":\n print(\"Error: WASENDER_API_KEY not set for channel example.\")\n return\n\n # For sending to channels, the SDK might have a specific helper or use the generic send.\n # This example assumes use of generic send() with ChannelTextMessage model as shown previously.\n # If a client.send_channel_text() helper exists, prefer that.\n\n async with create_async_wasender(api_key=local_api_key) as async_client:\n print(f\"\\nAttempting to send to WhatsApp Channel {channel_jid}...\")\n try:\n payload = ChannelTextMessage(\n to=channel_jid, \n message_type=\"text\", # Assuming model requires this\n text=text_message\n )\n # Using generic send method for channel messages as per original example\n result = await async_client.send(payload) \n \n print(f\"Message sent to channel successfully.\")\n if result.response: print(f\" Message ID: {result.response.message_id}, Status: {result.response.message}\")\n if result.rate_limit: print(f\"Rate limit: {result.rate_limit.remaining}/{result.rate_limit.limit}\")\n\n except WasenderAPIError as e:\n print(f\"API Error sending to channel: {e.message}\")\n except Exception as e:\n print(f\"An unexpected error: {e}\")\n\n# To run this example (ensure WASENDER_API_KEY and TEST_CHANNEL_JID are set):\n# async def main_run_channel_example():\n# test_channel_jid = os.getenv(\"TEST_CHANNEL_JID\", \"YOUR_CHANNEL_JID@newsletter\")\n# message = \"Hello Channel from SDK Example!\"\n# if test_channel_jid == \"YOUR_CHANNEL_JID@newsletter\":\n# print(\"Please set a valid TEST_CHANNEL_JID environment variable.\")\n# return\n# await send_to_channel_example(channel_jid=test_channel_jid, text_message=message)\n\n# if __name__ == \"__main__\":\n# asyncio.run(main_run_channel_example())\n```\n\n### 5. Handling Incoming Webhooks\n\nProcess real-time events from Wasender. The `handle_webhook_event` method is available on both sync and async clients, but it is an `async` method in both cases.\n\n- **Detailed Documentation & Examples:** [`docs/webhook.md`](./docs/webhook.md)\n\n```python\nimport os\nimport json # For pretty printing\n# Use the appropriate client factory\nfrom wasenderapi import create_sync_wasender, create_async_wasender \nfrom wasenderapi.webhook import WasenderWebhookEvent, WasenderWebhookEventType\nfrom wasenderapi.errors import SignatureVerificationError, WasenderAPIError \n\n# Client initialization for webhook handling:\n# Webhook secret is the key component here.\nwebhook_secret_from_env = os.getenv(\"WASENDER_WEBHOOK_SECRET\")\nif not webhook_secret_from_env:\n print(\"CRITICAL: WASENDER_WEBHOOK_SECRET environment variable is not set.\")\n # webhook_secret_from_env = \"YOUR_PLACEHOLDER_WEBHOOK_SECRET\" # For structure viewing\n\n# You can use either sync or async client to access handle_webhook_event.\n# However, since handle_webhook_event itself is async, using it in a purely\n# synchronous web framework like standard Flask requires special handling (e.g., asyncio.run_coroutine_threadsafe).\n# For an async framework (like FastAPI, Quart), you'd use the async client.\n\n# Example: Initialize a sync client for its config, but note the method is async.\nsync_client_for_webhook = create_sync_wasender(\n api_key=\"DUMMY_API_KEY_NOT_USED_FOR_WEBHOOK_LOGIC\", # API key not strictly needed for webhook verification logic\n webhook_secret=webhook_secret_from_env\n)\n\n# --- Conceptual Example: How to use with a web framework (e.g., Flask) ---\n# This illustrates the logic. Actual integration needs to handle the async nature\n# of handle_webhook_event within the sync framework.\n#\n# from flask import Flask, request, jsonify\n# import asyncio\n#\n# app = Flask(__name__)\n#\n# def run_async_in_sync(coro):\n# # A simple way to run an async function from sync code (for demonstration)\n# # In production, use framework-specific solutions or proper async event loop management.\n# loop = asyncio.new_event_loop()\n# asyncio.set_event_loop(loop)\n# return loop.run_until_complete(coro)\n#\n# @app.route(\"/wasender-webhook\", methods=[\"POST\"])\n# def flask_webhook_handler():\n# raw_body_bytes = request.get_data()\n# headers_dict = dict(request.headers)\n#\n# if not sync_client_for_webhook.webhook_secret:\n# print(\"Error: Webhook secret not configured.\")\n# return jsonify({\"error\": \"Webhook secret not configured\"}), 500\n#\n# try:\n# # handle_webhook_event is async, so needs to be run in an event loop\n# event: WasenderWebhookEvent = run_async_in_sync(\n# sync_client_for_webhook.handle_webhook_event(\n# request_body_bytes=raw_body_bytes, # Pass raw_body_bytes here\n# signature_header=headers_dict.get(WasenderWebhookEvent.SIGNATURE_HEADER) # Pass correct header\n# )\n# )\n# print(f\"Webhook Type: {event.event_type.value}\")\n# # ... (process event.data based on event.event_type) ...\n# return jsonify({\"status\": \"success\"}), 200\n# except SignatureVerificationError as e:\n# return jsonify({\"error\": \"Signature verification failed\"}), 400\n# except WasenderAPIError as e: # Handles other SDK errors like bad payload\n# return jsonify({\"error\": f\"Webhook processing error: {e.message}\"}), 400\n# except Exception as e:\n# return jsonify({\"error\": \"Internal server error\"}), 500\n\n# Dummy data for local testing (if __name__ == \"__main__\")\n# ... (The existing dummy data and call simulation can be kept but adapted ...)\n# ... ensure to call sync_client_for_webhook.handle_webhook_event ...\n\n# if __name__ == \"__main__\":\n# # ... (simulation code using sync_client_for_webhook.handle_webhook_event)\n# # Remember the async nature when simulating the call directly.\n# pass \n```\n\n### 6. Managing WhatsApp Sessions\n\nCreate, list, update, delete sessions, connect/disconnect, get QR codes, and check session status. Listing all sessions or creating new ones typically requires a `personal_access_token`.\n\n- **Detailed Documentation & Examples:** [`docs/sessions.md`](./docs/sessions.md)\n\n```python\nimport asyncio\nimport os\nfrom wasenderapi import create_async_wasender # Use factory for async client\nfrom wasenderapi.errors import WasenderAPIError\n\nasync def manage_whatsapp_sessions_example():\n local_api_key = os.getenv(\"WASENDER_API_KEY\", \"YOUR_SESSIONS_API_KEY\")\n # For listing all sessions or creating new ones, a personal access token is often needed.\n local_personal_access_token = os.getenv(\"WASENDER_PERSONAL_ACCESS_TOKEN\") \n\n if local_api_key == \"YOUR_SESSIONS_API_KEY\":\n print(\"Error: WASENDER_API_KEY not set for sessions example.\")\n return\n \n # Client for session-specific actions (uses api_key of that session)\n # Client for account-level session actions (uses personal_access_token)\n # We'll use one client, assuming personal_access_token is for listing/creating if needed,\n # and api_key for specific session operations if that's how the SDK/API is structured.\n # The create_async_wasender can take both.\n\n async with create_async_wasender(api_key=local_api_key, personal_access_token=local_personal_access_token) as async_client:\n if not local_personal_access_token:\n print(\"Warning: WASENDER_PERSONAL_ACCESS_TOKEN not set. Listing all sessions might fail or be restricted.\")\n\n print(\"\\nAttempting to list WhatsApp sessions...\")\n try:\n # get_all_whatsapp_sessions is the method for listing all sessions\n list_sessions_result = await async_client.get_all_whatsapp_sessions() \n\n if list_sessions_result.response and list_sessions_result.response.data is not None:\n sessions = list_sessions_result.response.data\n print(f\"Successfully fetched {len(sessions)} session(s).\")\n if sessions:\n for session_info in sessions:\n print(f\" Session ID: {session_info.session_id}, Status: {session_info.status.value if session_info.status else 'N/A'}\")\n else:\n print(\" No active sessions found for this account.\")\n else:\n print(\"No session data received.\")\n\n if list_sessions_result.rate_limit: print(f\"List Sessions Rate limit: {list_sessions_result.rate_limit.remaining}\")\n\n # Example for get_session_status (uses api_key of a specific session)\n # if sessions:\n # target_session_id = sessions[0].session_id # This is numeric ID from the list\n # # To get status, you typically use the API key associated with *that* session_id,\n # # which might be different from the local_api_key if it's a PAT.\n # # For this example, we assume async_client was initialized with the correct session's API key\n # # if we were to uncomment and run the following. Or, one might re-initialize a client\n # # specifically for that session if its API key is known.\n # print(f\"\\nAttempting to get status for session: {target_session_id}\")\n # try:\n # # Assuming local_api_key IS the key for this specific session if we call get_session_status\n # # If not, this call might not be correctly authorized.\n # status_result = await async_client.get_session_status(session_id=str(target_session_id)) # Ensure session_id is string if required\n # if status_result.response and status_result.response.data:\n # print(f\" Status for {target_session_id}: {status_result.response.data.status.value}\")\n # except WasenderAPIError as e_status:\n # print(f\" Error getting status for {target_session_id}: {e_status.message}\")\n\n except WasenderAPIError as e:\n print(f\"API Error managing sessions: {e.message}\")\n except Exception as e:\n print(f\"An unexpected error: {e}\")\n\n# To run this example (ensure WASENDER_API_KEY and optionally WASENDER_PERSONAL_ACCESS_TOKEN are set):\n# if __name__ == \"__main__\":\n# asyncio.run(manage_whatsapp_sessions_example())\n```\n\n## Advanced Topics\n\n### Configuring Retries\n\nThe SDK supports automatic retries, primarily for handling HTTP 429 (Too Many Requests) errors. This is configured via the `RetryConfig` object passed to `retry_options` during client initialization.\n\n```python\nfrom wasenderapi import create_sync_wasender, create_async_wasender\nfrom wasenderapi.models import RetryConfig\n\n# Configure retries: enable and set max retries\nretry_settings = RetryConfig(enabled=True, max_retries=3)\n\n# For synchronous client\nsync_client_with_retries = create_sync_wasender(\n api_key=\"YOUR_API_KEY\", \n retry_options=retry_settings\n)\n\n# For asynchronous client\nasync_client_with_retries = create_async_wasender(\n api_key=\"YOUR_API_KEY\", \n retry_options=retry_settings\n)\n\n# Example usage (sync)\n# try:\n# sync_client_with_retries.send_text(to=\"PHONE\", text_body=\"Test with retries\")\n# except WasenderAPIError as e:\n# print(f\"Failed after retries: {e}\")\n```\n\nBy default, retries are disabled.\n\n## Contributing\n\nContributions are welcome! Please feel free to submit issues, fork the repository, and create pull requests.\n\n## License\n\nThis SDK is released under the [MIT License](./LICENSE).\n",
"bugtrack_url": null,
"license": null,
"summary": "The official Python SDK for the Wasender API, allowing you to programmatically send WhatsApp messages, manage contacts, groups, sessions, and handle webhooks.",
"version": "0.3.4",
"project_urls": {
"Bug Tracker": "https://github.com/YonkoSam/wasenderapi-python/issues",
"Documentation": "https://github.com/YonkoSam/wasenderapi-python/tree/main/docs",
"Homepage": "https://github.com/YonkoSam/wasenderapi-python",
"Repository": "https://github.com/YonkoSam/wasenderapi-python"
},
"split_keywords": [
"api",
" chatbot",
" messaging",
" sdk",
" wasender",
" whatsapp"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "0ecd38697ba7bf47668f62065c3af55a55b543fd257d6b2ab52ed4c6316824a7",
"md5": "816d525ba15c22e4568b17c299eafe7b",
"sha256": "863f8d8f825685327d46a95422de8d5e3123d25e7f287a3f765470324ae51a73"
},
"downloads": -1,
"filename": "wasenderapi-0.3.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "816d525ba15c22e4568b17c299eafe7b",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 32918,
"upload_time": "2025-10-18T11:38:16",
"upload_time_iso_8601": "2025-10-18T11:38:16.366146Z",
"url": "https://files.pythonhosted.org/packages/0e/cd/38697ba7bf47668f62065c3af55a55b543fd257d6b2ab52ed4c6316824a7/wasenderapi-0.3.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "a276a75eeecbd1283b2ef0a6e52844012bfc498f20723af838252f44a710f3ba",
"md5": "da6f70b732b89b4aaa65a180a9a0e665",
"sha256": "de4c94c58b03db0aa9c5eb070206545365dfd5229e37fe72e268107c51058453"
},
"downloads": -1,
"filename": "wasenderapi-0.3.4.tar.gz",
"has_sig": false,
"md5_digest": "da6f70b732b89b4aaa65a180a9a0e665",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 84605,
"upload_time": "2025-10-18T11:38:17",
"upload_time_iso_8601": "2025-10-18T11:38:17.946618Z",
"url": "https://files.pythonhosted.org/packages/a2/76/a75eeecbd1283b2ef0a6e52844012bfc498f20723af838252f44a710f3ba/wasenderapi-0.3.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-18 11:38:17",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "YonkoSam",
"github_project": "wasenderapi-python",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [
{
"name": "requests",
"specs": [
[
">=",
"2.31.0"
]
]
},
{
"name": "pydantic",
"specs": [
[
">=",
"2.5.0"
]
]
},
{
"name": "pytest",
"specs": [
[
">=",
"7.4.0"
]
]
},
{
"name": "pytest-cov",
"specs": [
[
">=",
"4.1.0"
]
]
},
{
"name": "pytest-mock",
"specs": [
[
">=",
"3.12.0"
]
]
},
{
"name": "httpx",
"specs": [
[
">=",
"0.25.0"
]
]
}
],
"lcname": "wasenderapi"
}
JSON
