File: //home/arjun/projects/buyercall/buyercall/integrations/push_engage.py
import requests
import logging as log
log = log.getLogger(__name__)
class PushEngageAPI():
"""
Send Push Notifications using PushEngage API.
"""
request_timeout = 5 # Second(s)
request_retries = 3 # Integer
endpoint = None
api_key = None
# Valid parameters in PushEngage API
pushengage_api_params = [
'notification_url', 'notification_title', 'notification_message', 'image_url',
'include_segments', 'exclude_segments', 'include_countries', 'exclude_countries', 'include_states',
'include_cities', 'include_device_types', 'include_browsers', 'from_date', 'to_date', 'notification_expiry',
'notification_type', 'subscriber_hash', 'big_image_url', 'valid_from_utc'
]
# The notification event type which has a push notification
push_notification_event_types = [
'MISSED_CALL', 'VOICE_MAIL', 'TEXT_MESSAGE', 'FORM_SUBMISSION', 'PASSWORD_CHANGE'
]
def init_app(self, app):
"""
Set the endpoint and api key for the push engage api request.
:param app: flask app
:return None: None
"""
self.endpoint = app.config.get('PUSHENGAGE_ENDPOINT', 'https://api.pushengage.com/apiv1/notifications')
self.api_key = app.config.get('PUSHENGAGE_API_KEY', '871e2367-0809-4b22-8dd2-32475e4580f3')
return
def send_request(self, data, headers, retries, timeout=3):
try:
success = False
response = requests.post(url=self.endpoint, data=data, headers=headers, timeout=timeout)
if response:
success = True
retries -= 1
if not success and retries > 0:
self.send_request(data, headers, retries)
return response
except Exception as e:
log.error(f'{e}')
return
def send_push_notification(self, **kwargs):
"""
Send push notification
:param notification_url: Hyperlink for notification
:param notification_title: Title for notification
:param notification_message: Message dispalyed in notification body
:param profile_ids: User(s) who need to receive notifications
:return: json {'message': 'string', 'status': bool}
"""
response = None
notification_url = kwargs.get('notification_url', None)
notification_title = kwargs.get('notification_title', None)
notification_message = kwargs.get('notification_message', None)
profile_ids = kwargs.get('profile_ids', None)
if notification_url and notification_title and notification_message and profile_ids:
headers = {'api_key': self.api_key, 'Content-Type': 'application/x-www-form-urlencoded'}
data = {k:v for k, v in kwargs.items() if k in self.pushengage_api_params} # create dictionary values from params
for i, pid in enumerate(profile_ids):
data[f'profile_id[{i}]'] = str(pid)
response = self.send_request(data, headers, retries=self.request_retries, timeout=self.request_timeout)
return response.json() if response else {
'status': False,
'message': 'Bad Request. Required Parameters: notification_url, notification_title, notification_message and profile_id'
}
def schedule_push_notification(self, **kwargs):
"""
schedule push notification
:param notification_url: URL to open upon clicking on the push notification.
:param notification_title: Title of the push notification.
:param notification_message: The message to be displayed in the push notification.
:param notification_type: This defines the notification status, such as now, later or, draft. In this case, notification_type has always been "later".
:param valid_from_utc: This defines sent time of schedule notification. it is based on timezone of the configured account. This is valid only notification_type is later.
:return: json {'message': 'string', 'status': bool}
"""
return