This repo is made for the async version of the exponent-server-sdk-python made by Bahadır Araz.
pip install exponent-server-sdk-async
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.
importasyncioimportosimporthttpxfromloguruimportloggerfromdotenvimportload_dotenvfromexponent_server_sdk_asyncimport (
AsyncPushClient,
PushMessage,
DeviceNotRegisteredError,
PushTicketError,
PushServerError
)
# Logger setuplogger.add("debug.log", rotation="1 week", compression="zip")
load_dotenv()
logger.debug("Using Expo Token: {}", os.getenv('EXPO_TOKEN'))
asyncdefsend_push_message_to_multiple(tokens, message, extra, title, navigate_to=None):
asyncwithhttpx.AsyncClient(headers={
"Authorization": f"Bearer {os.getenv('EXPO_TOKEN')}",
"Accept": "application/json",
"Content-Type": "application/json",
}) assession:
push_client=AsyncPushClient(session=session)
push_messages= [
PushMessage(
to=token,
body=message,
data={'extra': extra, 'navigateTo': navigate_to},
badge=1,
title=title,
sound="default"
) fortokenintokens
]
try:
push_tickets=awaitpush_client.publish_multiple(push_messages)
forpush_ticketinpush_tickets:
ifpush_ticket.is_success():
logger.info("Notification sent successfully to: {}", push_ticket.push_message.to)
else:
logger.warning("Failed to send notification to: {}, Error: {}", push_ticket.push_message.to, push_ticket.message)
exceptPushServerErrorasexc:
logger.error("Push server error: {}", exc)
raiseexceptDeviceNotRegisteredErrorasexc:
logger.error("Device not registered error: {}", exc)
raiseexceptPushTicketErrorasexc:
logger.error("Push ticket error: {}", exc)
raiseasyncdefmain():
tokens= [
"ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]",
"ExponentPushToken[yyyyyyyyyyyyyyyyyyyyyy]",
"ExponentPushToken[zzzzzzzzzzzzzzzzzzzzzz]"
]
extra_parameters= {"key": "value"}
# Send a simple push message to multiple recipientsawaitsend_push_message_to_multiple(tokens, "Hello Multi-User", extra_parameters, "Greeting")
# Navigate to home data for multiple recipientsawaitsend_push_message_to_multiple(tokens, "Navigate to home", extra_parameters, "Home", "/(tabs)/(home)")
if__name__=="__main__":
asyncio.run(main())