from pyapns_client import APNSClient, IOSPayloadAlert, IOSPayload, IOSNotification, APNSDeviceException, APNSServerException, APNSProgrammingException, UnregisteredException
from concurrent.futures import ThreadPoolExecutor
import time
def send_notification(device_token):
try:
alert = IOSPayloadAlert(title='Title', body='Some message.')
payload = IOSPayload(alert=alert,badge=1,sound='default')
notification = IOSNotification(payload=payload, topic='topic')
client.push(notification=notification, device_token=device_token)
print(f'Successfully sent notification to {device_token}')
except UnregisteredException as e:
print(f'Device {device_token} is unregistered, compare timestamp {e.timestamp_datetime} and remove from db')
except APNSDeviceException:
print(f'Device {device_token} is potentially invalid, flag and remove from db after a few tries')
except APNSServerException:
print(f'Device {device_token}: try again later')
except APNSProgrammingException:
print(f'Device {device_token}: check your code and try again later')
# Your device tokens list
device_tokens = ['device_token1', 'device_token2']
# Initialize the APNS client
client = APNSClient(mode=APNSClient.MODE_DEV, root_cert_path=None, auth_key_path='auth_key', auth_key_id='auth_key_id', team_id='team_id')
start_time = time.time()
try:
with ThreadPoolExecutor(max_workers=4) as executor: # You can adjust the number of workers as needed
executor.map(send_notification, device_tokens)
finally:
client.close()
end_time = time.time()
print (end_time - start_time)