pywebpush

You can install pywebpush from PyPI. To install it locally for development, use a virtual environment and install in editable mode:

python -m venv venv
venv/bin/pip install --editable .

Microsoft Windows requires specific non-standard headers for push notifications to appear correctly. As of April 2024, X-WNS-Type is required. You can provide these via a JSON file using the --head flag in the CLI or the headers parameter in the Python API.

Example windows_headers.json:

{"X-WNS-Type":"wns/toast", "TTL":600, "Content-Type": "text/xml"}

CLI usage with headers:

pywebpush --data stuff_to_send.xml \
   --info edge_user_info.json \
   --head windows_headers.json \
   --claims vapid_claims.json

To use VAPID (Voluntary Application Server Identification), provide a vapid_private_key and vapid_claims to webpush() or webpush_async().

• vapid_private_key: Can be a py_vapid.Vapid instance, a path to a PEM file, or an encoded string.
• vapid_claims: A dictionary of claims. The sub (subject) claim is required. The library will automatically calculate the aud (audience) from the endpoint URL and set an exp (expiry) if not provided.

The WebPusher class provides two primary methods:

.send(data, headers={}, ttl=0, reg_id="", content_encoding="aes128gcm", curl=False, timeout=None)

Sends the data to the push server.

• data: Binary string of data to send.
• headers: A dict of additional HTTP headers.
• ttl: Message Time To Live in seconds.
• reg_id: GCM registration ID (extracted from endpoint if not provided).
• content_encoding: ECE content encoding type (defaults to "aes128gcm").
• curl: If True, does not execute the POST but returns a curl command for debugging. It writes encrypted content to encrypted.data.
• timeout: Request timeout (compatible with requests library).

.encode(data, content_encoding="aes128gcm")

Encodes the data for future use. Returns a WebPushException if data is empty (raises NoData).

If you need to reuse a subscription or require more granular control, instantiate a WebPusher object with the subscription_info.

from pywebpush import WebPusher

wp = WebPusher(subscription_info)
# Use wp.send(...) or wp.encode(...)

A standalone CLI tool is available in the ./bin directory for testing push interfaces without writing code. It requires a data file and a subscription info JSON file.

./bin/pywebpush --data stuff_to_send.data --info subscription.info

The webpush_async() function is the asynchronous version of webpush(). It is suitable for asyncio environments and allows for connection reuse by passing an optional aiohttp_session. Like the synchronous version, it raises WebPushException on non-success responses.

from pywebpush import webpush_async
import asyncio

async def send_notification():
    response = await webpush_async(
        subscription_info={
            "endpoint": "https://push.example.com/v1/abcd",
            "keys": {
                "p256dh": "0123abcd...",
                "auth": "001122..."
            }
        },
        data="Mary had a little lamb, with a nice mint jelly",
        vapid_private_key="path/to/key.pem",
        vapid_claims={"sub": "YourNameHere@example.com"}
    )

asyncio.run(send_notification())

When a push request fails or the subscription information is invalid, the library raises a WebPushException. This exception includes a message and, if available, the response object (from requests or aiohttp) which contains the error body from the push service.

try:
    webpush(...)
except WebPushException as e:
    print(f"Error: {e.message}")
    if e.response is not None:
        print(f"Response body: {e.response.text}")

The webpush() function is a high-level, synchronous solution to encode and send data to a Web Push endpoint. It handles encryption and optional VAPID authentication in a single call. If the request fails (status code > 202), it raises a WebPushException.

from pywebpush import webpush

webpush(
    subscription_info={
        "endpoint": "https://push.example.com/v1/abcd",
        "keys": {
            "p256dh": "0123abcd...",
            "auth": "001122..."
        }
    },
    data="Mary had a little lamb, with a nice mint jelly",
    vapid_private_key="path/to/key.pem",
    vapid_claims={"sub": "YourNameHere@example.com"}
)

For more granular control, use the WebPusher class. You initialize it with the client's subscription_info and can optionally provide a requests.Session (for synchronous) or aiohttp.ClientSession (for asynchronous) to optimize requests.

Key methods:

• encode(data, content_encoding): Encrypts the data block.
• send(*args, **kwargs): Encodes and sends the data synchronously.
• send_async(*args, **kwargs): Encodes and sends the data asynchronously.
• as_curl(...): Returns a curl command string for debugging, which writes the encoded data to a local encrypted.data file.

# Example of manual WebPusher usage
subscription_info = {
    "endpoint": "https://push.server.com/...",
    "keys": {"auth": "...", "p256dh": "..."}
}

pusher = WebPusher(subscription_info)
data = "Hello World"
pusher.send(data)