async-expo-push-notifications 2.2.0

Expo Server SDK for Python with async/await support, Pydantic models, and full type hints

exponent-server-sdk-python

This repo is maintained by Expo's awesome community :heart_eyes:! So, if you have problems with the code in this repository, please feel free to open an issue, and make a PR. Thanks!

Installation

pip install exponent_server_sdk

Requirements

• Python 3.8+
• Type-safe with Pydantic models
• Async support with httpx
• Synchronous support with requests (backward compatible)

Usage

Use to send push notifications to Exponent Experiences from a Python server.

Full documentation on the API is available if you want to dive into the details.

Async/Await Usage (Recommended for modern applications)

import asyncio
from exponent_server_sdk import AsyncPushClient, PushMessage

async def send_notification():
    async with AsyncPushClient() as client:
        message = PushMessage(
            to="ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]",
            title="Hello",
            body="World!",
            data={"extra": "data"}
        )
        ticket = await client.publish(message)
        ticket.validate_response()

asyncio.run(send_notification())

Synchronous Usage (Backward Compatible)

Here's an example on how to use this with retries and reporting via pyrollbar.

from exponent_server_sdk import (
    DeviceNotRegisteredError,
    PushClient,
    PushMessage,
    PushServerError,
    PushTicketError,
)
import os
import requests
from requests.exceptions import ConnectionError, HTTPError

# Optionally providing an access token within a session if you have enabled push security
session = requests.Session()
session.headers.update(
    {
        "Authorization": f"Bearer {os.getenv('EXPO_TOKEN')}",
        "accept": "application/json",
        "accept-encoding": "gzip, deflate",
        "content-type": "application/json",
    }
)

# Basic arguments. You should extend this function with the push features you
# want to use, or simply pass in a `PushMessage` object.
def send_push_message(token, message, extra=None):
    try:
        response = PushClient(session=session).publish(
            PushMessage(to=token,
                        body=message,
                        data=extra))
    except PushServerError as exc:
        # Encountered some likely formatting/validation error.
        rollbar.report_exc_info(
            extra_data={
                'token': token,
                'message': message,
                'extra': extra,
                'errors': exc.errors,
                'response_data': exc.response_data,
            })
        raise
    except (ConnectionError, HTTPError) as exc:
        # Encountered some Connection or HTTP error - retry a few times in
        # case it is transient.
        rollbar.report_exc_info(
            extra_data={'token': token, 'message': message, 'extra': extra})
        raise self.retry(exc=exc)

    try:
        # We got a response back, but we don't know whether it's an error yet.
        # This call raises errors so we can handle them with normal exception
        # flows.
        response.validate_response()
    except DeviceNotRegisteredError:
        # Mark the push token as inactive
        from notifications.models import PushToken
        PushToken.objects.filter(token=token).update(active=False)
    except PushTicketError as exc:
        # Encountered some other per-notification error.
        rollbar.report_exc_info(
            extra_data={
                'token': token,
                'message': message,
                'extra': extra,
                'push_response': exc.push_response._asdict(),
            })
        raise self.retry(exc=exc)

Features

🚀 Async/Await Support

Modern async/await syntax for high-performance applications:

async with AsyncPushClient() as client:
    tickets = await client.publish_multiple(messages)

🔒 Type Safety with Pydantic

All models use Pydantic for validation and type safety:

# Automatic validation
message = PushMessage(
    to="ExponentPushToken[xxx]",
    priority="high",  # Validates against allowed values
    richContent={"image": "https://example.com/img.png"}  # NEW!
)

💉 Dependency Injection

Inject custom HTTP clients for testing or custom behavior:

import httpx

custom_client = httpx.AsyncClient(
    headers={"Authorization": f"Bearer {token}"}
)
push_client = AsyncPushClient(http_client=custom_client)

📸 Rich Content Support

Now includes the richContent field for image notifications:

message = PushMessage(
    to=token,
    title="Check this out!",
    body="Beautiful image notification",
    richContent={"image": "https://example.com/image.jpg"}
)

Examples

Check out the examples/ directory for more:

• async_example.py - Async/await usage
• sync_example.py - Synchronous usage
• dependency_injection_example.py - DI patterns

Migration Guide

From v2.1.x to v2.2.0

The library is fully backward compatible. Existing code will continue to work without changes:

# Old code still works!
from exponent_server_sdk import PushClient, PushMessage

client = PushClient()
ticket = client.publish(PushMessage(to=token, body="Hello"))

New Features You Can Adopt

1. Use Pydantic models - Your PushMessage objects now have validation
2. Try async/await - Use AsyncPushClient for better performance
3. Add richContent - Include images in your notifications
4. Inject dependencies - Pass custom HTTP clients for testing

Type Hints

All classes and methods now include full type hints:

from exponent_server_sdk import AsyncPushClient, PushMessage, PushTicket
from typing import List

async def send_many(messages: List[PushMessage]) -> List[PushTicket]:
    async with AsyncPushClient() as client:
        return await client.publish_multiple(messages)

Contributing

Issues and pull requests are welcome! See CONTRIBUTING.md for guidelines.

License

MIT