Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
for target in list(targets):
info = self.registrations.get(target)
try:
info = REGISTER_SCHEMA(info)
except vol.Invalid:
_LOGGER.error(
"%s is not a valid HTML5 push notification" " target", target
)
continue
payload[ATTR_DATA][ATTR_JWT] = add_jwt(
timestamp,
target,
payload[ATTR_TAG],
info[ATTR_SUBSCRIPTION][ATTR_KEYS][ATTR_AUTH],
)
webpusher = WebPusher(info[ATTR_SUBSCRIPTION])
if self._vapid_prv and self._vapid_email:
vapid_headers = create_vapid_headers(
self._vapid_email, info[ATTR_SUBSCRIPTION], self._vapid_prv
)
vapid_headers.update({"urgency": priority, "priority": priority})
response = webpusher.send(
data=json.dumps(payload), headers=vapid_headers, ttl=ttl
)
else:
# Only pass the gcm key if we're actually using GCM
# If we don't, notifications break on FireFox
gcm_key = (
self._gcm_key
if "googleapis.com" in info[ATTR_SUBSCRIPTION][ATTR_ENDPOINT]
else None
)
Desc: Web Push notifications for Chrome and FireFox
Installation:
pip install 'pywebpush>=0.4.0'
"""
subscription_info = kwargs.pop('subscription_info')
payload = {
"title": kwargs.pop("event"),
"body": message,
"url": kwargs.pop("push_url", None)
}
payload.update(kwargs)
wp = WebPusher(subscription_info)
response = wp.send(
dumps(payload), gcm_key=settings.GCM_KEY,
ttl=kwargs.pop("ttl", 60))
if not response.ok or (
response.text and loads(response.text).get("failure") > 0):
raise GCMError(response.text)
return True
for registration in registrations:
user_subscriptions, count = request.registry.storage.get_all(
collection_id=SUBSCRIPTION_COLLECTION_ID,
parent_id=registration['id'])
subscriptions += user_subscriptions
for subscription in subscriptions:
# Remove the subscription id if present.
subscription.pop('id', None)
# Remove the subscription last_modified value if present.
subscription.pop('last_modified')
data = request.validated.get('data')
if data:
data = canonical_json(data)
push_initialize = WebPusher(subscription)
push_initialize.send(data=data, ttl=15)
request.response.status = 202
return {"code": 202, "message": "Accepted"}
def process_record(self, new, old=None):
new = super(Subscription, self).process_record(new, old)
try:
WebPusher(new)
except WebPushException as e:
raise http_error(HTTPBadRequest(),
errno=ERRORS.INVALID_PARAMETERS,
message='Invalid subscription: %s' % e)
return new