Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, conf, router_conf, metrics):
"""Create a new FCM router and connect to FCM"""
self.conf = conf
self.router_conf = router_conf
self.metrics = metrics
self.min_ttl = router_conf.get("ttl", 60)
self.dryRun = router_conf.get("dryrun", False)
self.collapseKey = router_conf.get("collapseKey", "webpush")
self.senderID = router_conf.get("senderID")
self.auth = router_conf.get("auth")
self._base_tags = ["platform:fcm"]
try:
self.fcm = pyfcm.FCMNotification(api_key=self.auth)
except Exception as e:
self.log.error("Could not instantiate FCM {ex}",
ex=e)
raise IOError("FCM Bridge not initiated in main")
self.log.debug("Starting FCM router...")
"databaseURL": databaseURL,
"storageBucket": ""
}
firebase = pyrebase.initialize_app(config_pyrebase)
mAuth = firebase.auth()
user = mAuth.sign_in_with_email_and_password(mail, password)
userID = mAuth.get_account_info(user['idToken'])
userID = userID["users"][0]["localId"]
db = firebase.database()
db_user = db.child("users").child(userID).get(user['idToken'])
deviceID = db_user.val()
push_service = FCMNotification(api_key=apiKeyFCM)
conn = sqlite3.connect('/opt/kripton-guard/network.db')
def createTables(conn):
#Create db table if it's not exist
conn.execute("CREATE TABLE mac_ip_addresses (ID INTEGER PRIMARY KEY AUTOINCREMENT, macAddress varchar(17) UNIQUE NOT NULL, ipAddress varchar(15) NOT NULL, comment varchar(50) )")
def showDevices():
#Shows devices in whitelist
print "\n========== Your Whitelist ==========\n Mac Address IP Address "
query = "SELECT macAddress,ipAddress FROM mac_ip_addresses;"
result = conn.execute(query)
for row in result:
print row[0] + " " + row[1] + "\n"
print "==================================="
from pyfcm import FCMNotification
# fcm 푸쉬 메시지는 data message를 전송할 수 있다
fcm = FCMNotification(api_key='')
data = {'key1': 123, 'key2': 1234}
fcm.notify_single_device(data_message=data)
click_action=None,
badge=None,
color=None,
tag=None,
body_loc_key=None,
body_loc_args=None,
title_loc_key=None,
title_loc_args=None,
content_available=None,
timeout=5,
extra_notification_kwargs=None,
extra_kwargs={}):
if api_key is None:
api_key = SETTINGS.get("FCM_SERVER_KEY")
push_service = FCMNotification(api_key=api_key, json_encoder=json_encoder)
result = push_service.notify_topic_subscribers(
topic_name=topic_name,
message_body=message_body,
message_title=message_title,
message_icon=message_icon,
sound=sound,
condition=condition,
collapse_key=collapse_key,
delay_while_idle=delay_while_idle,
time_to_live=time_to_live,
restricted_package_name=restricted_package_name,
low_priority=low_priority,
dry_run=dry_run,
data_message=data_message,
click_action=click_action,
badge=badge,
def _get_push_service():
return FCMNotification(api_key=_get_fcm_key())
import logging
from django.conf import settings
from pyfcm import FCMNotification
from foodsaving.subscriptions.models import PushSubscription
logger = logging.getLogger(__name__)
fcm = None
if hasattr(settings, 'FCM_SERVER_KEY'):
fcm = FCMNotification(api_key=settings.FCM_SERVER_KEY)
else:
logger.warning('Please configure FCM_SERVER_KEY in your settings to use push messaging')
def notify_multiple_devices(**kwargs):
"""
Send a message to multiple devices.
A simple wrapper of pyfcm's notify_multiple_devices.
See https://github.com/olucurious/PyFCM/blob/master/pyfcm/fcm.py for more details on options, etc.
"""
if fcm is None:
return None
response = fcm.notify_multiple_devices(**kwargs)
def __init__(self, config):
self.config = config
self.api_key = self.config.get('api_key')
# FCM guarantees best-effort delivery with TTL 0
self.ttl = self.config.get('ttl', 0)
self.timeout = self.config.get('timeout', 10)
self.default_notification = self.config.get('notification_title')
self.proxy = None
if 'proxy' in self.config:
host = self.config['proxy']['host']
port = self.config['proxy']['port']
self.proxy = {'http': 'http://%s:%s' % (host, port),
'https': 'https://%s:%s' % (host, port)}
self.client = FCMNotification(api_key=self.api_key, proxy_dict=self.proxy)
def send_push_notif(title, body):
# Generate push service object
push_service = FCMNotification(api_key=Constants.PUSH_ADDR)
# Send push notif. to android and the notify_single_device returns result.
ret = push_service.notify_single_device(registration_id=Constants.REG_ID, message_title=title, message_body=body)
print(ret)
timeout (int, optional): set time limit for the request
Returns:
:dict:`multicast_id(long), success(int), failure(int),
canonical_ids(int), results(list)`:
Response from FCM server.
Raises:
AuthenticationError: If :attr:`api_key` is not set or provided or there
is an error authenticating the sender.
FCMServerError: Internal server error or timeout error on Firebase cloud
messaging server
InvalidDataError: Invalid data provided
InternalPackageError: Mostly from changes in the response of FCM,
contact the project owner to resolve the issue
"""
push_service = FCMNotification(
api_key=SETTINGS.get("FCM_SERVER_KEY") if api_key is None else api_key,
json_encoder=json_encoder,
)
return push_service.single_device_data_message(
registration_id=registration_id,
condition=condition,
collapse_key=collapse_key,
delay_while_idle=delay_while_idle,
time_to_live=time_to_live,
restricted_package_name=restricted_package_name,
low_priority=low_priority,
dry_run=dry_run,
data_message=data_message,
content_available=content_available,
timeout=timeout,
extra_notification_kwargs=extra_notification_kwargs,