319 lines
11 KiB
Python
319 lines
11 KiB
Python
from collections import namedtuple
|
|
import json
|
|
|
|
|
|
class PushResponseError(Exception):
|
|
"""Base class for all push reponse errors"""
|
|
|
|
def __init__(self, push_response):
|
|
if push_response.message:
|
|
self.message = push_response.message
|
|
else:
|
|
self.message = 'Unknown push response error'
|
|
super(PushResponseError, self).__init__(self.message)
|
|
|
|
self.push_response = push_response
|
|
|
|
|
|
class DeviceNotRegisteredError(PushResponseError):
|
|
"""Raised when the push token is invalid
|
|
|
|
To handle this error, you should stop sending messages to this token.
|
|
"""
|
|
pass
|
|
|
|
|
|
class MessageTooBigError(PushResponseError):
|
|
"""Raised when the notification was too large.
|
|
|
|
On Android and iOS, the total payload must be at most 4096 bytes.
|
|
"""
|
|
pass
|
|
|
|
|
|
class MessageRateExceededError(PushResponseError):
|
|
"""Raised when you are sending messages too frequently to a device
|
|
|
|
You should implement exponential backoff and slowly retry sending messages.
|
|
"""
|
|
pass
|
|
|
|
|
|
class PushServerError(Exception):
|
|
"""Raised when the push token server is not behaving as expected
|
|
|
|
For example, invalid push notification arguments result in a different
|
|
style of error. Instead of a "data" array containing errors per
|
|
notification, an "error" array is returned.
|
|
|
|
{"errors": [
|
|
{"code": "API_ERROR",
|
|
"message": "child \"to\" fails because [\"to\" must be a string]. \"value\" must be an array."
|
|
}
|
|
]}
|
|
"""
|
|
|
|
def __init__(self, message, response, response_data=None, errors=None):
|
|
self.message = message
|
|
self.response = response
|
|
self.response_data = response_data
|
|
self.errors = errors
|
|
super(PushServerError, self).__init__(self.message)
|
|
|
|
|
|
class PushMessage(namedtuple('PushMessage', [
|
|
'to', 'data', 'title', 'body', 'sound', 'ttl', 'expiration',
|
|
'priority', 'badge', 'channel_id'])):
|
|
"""An object that describes a push notification request.
|
|
|
|
You can override this class to provide your own custom validation before
|
|
sending these to the Exponent push servers. You can also override the
|
|
get_payload function itself to take advantage of any hidden or new
|
|
arguments before this library updates upstream.
|
|
|
|
Args:
|
|
to: A token of the form ExponentPushToken[xxxxxxx]
|
|
data: A dict of extra data to pass inside of the push notification.
|
|
The total notification payload must be at most 4096 bytes.
|
|
title: The title to display in the notification. On iOS, this is
|
|
displayed only on Apple Watch.
|
|
body: The message to display in the notification.
|
|
sound: A sound to play when the recipient receives this
|
|
notification. Specify "default" to play the device's default
|
|
notification sound, or omit this field to play no sound.
|
|
ttl: The number of seconds for which the message may be kept around
|
|
for redelivery if it hasn't been delivered yet. Defaults to 0.
|
|
expiration: UNIX timestamp for when this message expires. It has
|
|
the same effect as ttl, and is just an absolute timestamp
|
|
instead of a relative one.
|
|
priority: Delivery priority of the message. 'default', 'normal',
|
|
and 'high' are the only valid values.
|
|
badge: An integer representing the unread notification count. This
|
|
currently only affects iOS. Specify 0 to clear the badge count.
|
|
channel_id: ID of the Notification Channel through which to display
|
|
this notification on Android devices.
|
|
|
|
"""
|
|
def get_payload(self):
|
|
# Sanity check for invalid push token format.
|
|
if not PushClient.is_exponent_push_token(self.to):
|
|
raise ValueError('Invalid push token')
|
|
|
|
# There is only one required field.
|
|
payload = {
|
|
'to': self.to,
|
|
}
|
|
|
|
# All of these fields are optional.
|
|
if self.data is not None:
|
|
payload['data'] = self.data
|
|
if self.title is not None:
|
|
payload['title'] = self.title
|
|
if self.body is not None:
|
|
payload['body'] = self.body
|
|
if self.sound is not None:
|
|
payload['sound'] = self.sound
|
|
if self.ttl is not None:
|
|
payload['ttl'] = self.ttl
|
|
if self.expiration is not None:
|
|
payload['expiration'] = self.expiration
|
|
if self.priority is not None:
|
|
payload['priority'] = self.priority
|
|
if self.badge is not None:
|
|
payload['badge'] = self.badge
|
|
if self.channel_id is not None:
|
|
payload['channelId'] = self.channel_id
|
|
return payload
|
|
|
|
|
|
# Allow optional arguments for PushMessages since everything but the `to` field
|
|
# is optional. Unfortunately namedtuples don't allow for an easy way to create
|
|
# a required argument at the contructor level right now.
|
|
PushMessage.__new__.__defaults__ = (None,) * len(PushMessage._fields)
|
|
|
|
|
|
class PushResponse(namedtuple('PushResponse', [
|
|
'push_message', 'status', 'message', 'details'])):
|
|
"""Wrapper class for a push notification response.
|
|
|
|
A successful single push notification:
|
|
{'status': 'ok'}
|
|
|
|
An invalid push token
|
|
{'status': 'error',
|
|
'message': '"adsf" is not a registered push notification recipient'}
|
|
"""
|
|
# Known status codes
|
|
ERROR_STATUS = 'error'
|
|
SUCCESS_STATUS = 'ok'
|
|
|
|
# Known error strings
|
|
ERROR_DEVICE_NOT_REGISTERED = 'DeviceNotRegistered'
|
|
ERROR_MESSAGE_TOO_BIG = 'MessageTooBig'
|
|
ERROR_MESSAGE_RATE_EXCEEDED = 'MessageRateExceeded'
|
|
|
|
def is_success(self):
|
|
"""Returns True if this push notification successfully sent."""
|
|
return self.status == PushResponse.SUCCESS_STATUS
|
|
|
|
def validate_response(self):
|
|
"""Raises an exception if there was an error. Otherwise, do nothing.
|
|
|
|
Clients should handle these errors, since these require custom handling
|
|
to properly resolve.
|
|
"""
|
|
if self.is_success():
|
|
return
|
|
|
|
# Handle the error if we have any information
|
|
if self.details:
|
|
error = self.details.get('error', None)
|
|
|
|
if error == PushResponse.ERROR_DEVICE_NOT_REGISTERED:
|
|
raise DeviceNotRegisteredError(self)
|
|
elif error == PushResponse.ERROR_MESSAGE_TOO_BIG:
|
|
raise MessageTooBigError(self)
|
|
elif error == PushResponse.ERROR_MESSAGE_RATE_EXCEEDED:
|
|
raise MessageRateExceededError(self)
|
|
|
|
# No known error information, so let's raise a generic error.
|
|
raise PushResponseError(self)
|
|
|
|
|
|
class PushClient(object):
|
|
"""Exponent push client
|
|
|
|
See full API docs at https://docs.expo.io/versions/latest/guides/push-notifications.html#http2-api
|
|
"""
|
|
DEFAULT_HOST = "https://exp.host"
|
|
DEFAULT_BASE_API_URL = "/--/api/v2"
|
|
|
|
def __init__(self, host=None, api_url=None):
|
|
"""Construct a new PushClient object.
|
|
|
|
Args:
|
|
host: The server protocol, hostname, and port.
|
|
api_url: The api url at the host.
|
|
"""
|
|
self.host = host
|
|
if not self.host:
|
|
self.host = PushClient.DEFAULT_HOST
|
|
|
|
self.api_url = api_url
|
|
if not self.api_url:
|
|
self.api_url = PushClient.DEFAULT_BASE_API_URL
|
|
|
|
@classmethod
|
|
def is_exponent_push_token(cls, token):
|
|
"""Returns `True` if the token is an Exponent push token"""
|
|
import six
|
|
|
|
return (
|
|
isinstance(token, six.string_types) and
|
|
token.startswith('ExponentPushToken'))
|
|
|
|
def _publish_internal(self, push_messages):
|
|
"""Send push notifications
|
|
|
|
The server will validate any type of syntax errors and the client will
|
|
raise the proper exceptions for the user to handle.
|
|
|
|
Each notification is of the form:
|
|
{
|
|
'to': 'ExponentPushToken[xxx]',
|
|
'body': 'This text gets display in the notification',
|
|
'badge': 1,
|
|
'data': {'any': 'json object'},
|
|
}
|
|
|
|
Args:
|
|
push_messages: An array of PushMessage objects.
|
|
"""
|
|
# Delayed import because this file is immediately read on install, and
|
|
# the requests library may not be installed yet.
|
|
import requests
|
|
|
|
response = requests.post(
|
|
self.host + self.api_url + '/push/send',
|
|
data=json.dumps([pm.get_payload() for pm in push_messages]),
|
|
headers={
|
|
'accept': 'application/json',
|
|
'accept-encoding': 'gzip, deflate',
|
|
'content-type': 'application/json',
|
|
}
|
|
)
|
|
|
|
# Let's validate the response format first.
|
|
try:
|
|
response_data = response.json()
|
|
except ValueError:
|
|
# The response isn't json. First, let's attempt to raise a normal
|
|
# http error. If it's a 200, then we'll raise our own error.
|
|
response.raise_for_status()
|
|
|
|
raise PushServerError('Invalid server response', response)
|
|
|
|
# If there are errors with the entire request, raise an error now.
|
|
if 'errors' in response_data:
|
|
raise PushServerError(
|
|
'Request failed',
|
|
response,
|
|
response_data=response_data,
|
|
errors=response_data['errors'])
|
|
|
|
# We expect the response to have a 'data' field with the responses.
|
|
if 'data' not in response_data:
|
|
raise PushServerError(
|
|
'Invalid server response',
|
|
response,
|
|
response_data=response_data)
|
|
|
|
# Use the requests library's built-in exceptions for any remaining 4xx
|
|
# and 5xx errors.
|
|
response.raise_for_status()
|
|
|
|
# Sanity check the response
|
|
if len(push_messages) != len(response_data['data']):
|
|
raise PushServerError(
|
|
('Mismatched response length. Expected %d %s but only '
|
|
'received %d' % (
|
|
len(push_messages),
|
|
'receipt' if len(push_messages) == 1 else 'receipts',
|
|
len(response_data['data']))),
|
|
response,
|
|
response_data=response_data)
|
|
|
|
# At this point, we know it's a 200 and the response format is correct.
|
|
# Now let's parse the responses per push notification.
|
|
receipts = []
|
|
for i, receipt in enumerate(response_data['data']):
|
|
receipts.append(PushResponse(
|
|
push_message=push_messages[i],
|
|
# If there is no status, assume error.
|
|
status=receipt.get('status', PushResponse.ERROR_STATUS),
|
|
message=receipt.get('message', ''),
|
|
details=receipt.get('details', None)))
|
|
|
|
return receipts
|
|
|
|
def publish(self, push_message):
|
|
"""Sends a single push notification
|
|
|
|
Args:
|
|
push_message: A single PushMessage object.
|
|
|
|
Returns:
|
|
A PushResponse object which contains the results.
|
|
"""
|
|
return self.publish_multiple([push_message])[0]
|
|
|
|
def publish_multiple(self, push_messages):
|
|
"""Sends multiple push notifications at once
|
|
|
|
Args:
|
|
push_messages: An array of PushMessage objects.
|
|
|
|
Returns:
|
|
An array of PushResponse objects which contains the results.
|
|
"""
|
|
return self._publish_internal(push_messages)
|