Ok, I am beyond frustrated at this point. I had given up on this back when I originally posted this but am now at a point where I really need this to work. I have gone through multiple other threads, and I simply cannot solve this issue. Ok, ok, I know, I need to calm down… Goosfraba…
Ok, here is what I have tried.
FORM CODE
import anvil.server
import anvil.tables as tables
import anvil.tables.query as q
import anvil.users
from anvil import *
from anvil.tables import app_tables
from ..FirebaseClient import ProjectApp, ProjectConfig, PushNotificationMessage
from ._anvil_designer import Form1Template
PUBLIC_VAPID_KEY = "BIWD7o3v3BkSxnUqiO----------K0Vgo_oCvDzRUWxUYXzYUBgaroso"
class Form1(Form1Template):
def __init__(self, **properties):
# Set Form properties and Data Bindings.
self.init_components(**properties)
# Any code you write here will run when the form opens.
self.label_ip.text = anvil.server.call("get_ip")
def initialize_app(self):
# Define your Firebase project configuration
project_config = ProjectConfig.from_dict(
{
"apiKey": "",
"authDomain": "",
"projectId": "",
"storageBucket": "",
"messagingSenderId": "",
"appId": "",
"measurementId": "",
}
) # I've removed the details here, but they are in my code
# Initialize ProjectApp with the configuration
self.project_app = ProjectApp(project_config, PUBLIC_VAPID_KEY, self.message_hadler)
self.project_app.initialize_app()
self.label_token.text = self.project_app.device_token
def message_hadler(self, payload, **event_args):
print(f"Payload: {payload}")
def button_1_click(self, **event_args):
"""This method is called when the button is clicked."""
# Retrieve the device token
device_token = (
self.project_app.device_token
) # Assuming the token is stored in the ProjectApp instance
# Create a PushNotificationMessage
notification_message = PushNotificationMessage(
title="Special Offer Just for You again!",
body="Click here to discover amazing deals available for a limited time.",
icon="https://w7.pngwing.com/pngs/818/522/png-transparent-special-offer-poster-computer-icons-discounts-and-allowances-icon-design-price-tag-miscellaneous-text-logo.png",
url_to_open_on_click="https://utter-dearest-mountain-goat.anvil.app",
)
# Send the notification
anvil.server.call(
"send_test_notification", # Ensure this is the correct server function name
push_notification_message=notification_message,
device_token=device_token,
)
# Add any additional logic as needed
def form_show(self, **event_args):
"""This method is called when the form is shown on the page"""
# Initialize Firebase app
self.initialize_app()
FIREBASE CLIENT CODE
import anvil.server
import anvil.tables as tables
import anvil.tables.query as q
import anvil.users
from anvil import *
from anvil.tables import app_tables
class ProjectConfig:
def __init__(
self,
api_key: str,
auth_domain: str,
project_id: str,
storage_bucket: str,
messaging_sender_id: str,
app_id: str,
measurement_id: str,
):
self.api_key = api_key
self.auth_domain = auth_domain
self.project_id = project_id
self.storage_bucket = storage_bucket
self.messaging_sender_id = messaging_sender_id
self.app_id = app_id
self.measurement_id = measurement_id
@property
def as_dict(self):
return {
"apiKey": self.api_key,
"authDomain": self.auth_domain,
"projectId": self.project_id,
"storageBucket": self.storage_bucket,
"messagingSenderId": self.messaging_sender_id,
"appId": self.app_id,
"measurementId": self.measurement_id,
}
@classmethod
def from_dict(cls, d: dict):
return cls(
api_key=d["apiKey"],
auth_domain=d["authDomain"],
project_id=d["projectId"],
storage_bucket=d["storageBucket"],
messaging_sender_id=d["messagingSenderId"],
app_id=d["appId"],
measurement_id=d["measurementId"],
)
class ProjectApp:
def __init__(
self,
project_config: ProjectConfig,
public_vapid_key: str = None,
message_handler: callable = None,
):
self.messaging = None
self.device_token = None
self.project_config = project_config
self.public_vapid_key = public_vapid_key
self.message_handler = message_handler
def initialize_app(self):
from anvil.js.window import firebase
firebase.initializeApp(self.project_config.as_dict)
print(f"Is Messaging Supported: {firebase.messaging.isSupported()}")
self.messaging = firebase.messaging()
self.register_service_worker()
self.request_notification_permission()
self.retrieve_and_save_device_token()
self.setup_token_refresh_handler()
self.setup_message_handler()
def register_service_worker(self):
from anvil.js import await_promise
from anvil.js.window import navigator
# Assuming there's a service worker file at the specified path
try:
# Register the service worker
registration = await_promise(
navigator.serviceWorker.register(
# _/theme/service-worker.js
f"{anvil.server.get_app_origin()}/_/theme/service-worker.js"
)
)
self.set_service_worker(registration)
except Exception as e:
print(f"Error registering service worker: {e}")
def set_service_worker(self, registration):
self.messaging.useServiceWorker(registration)
def request_notification_permission(self):
from anvil.js import await_promise
from anvil.js.window import window
# Request user permission for notifications
try:
permission = await_promise(window.Notification.requestPermission())
self.handle_permission(permission)
except Exception as e:
print(f"Error requesting notification permission: {e}")
def handle_permission(self, permission):
if permission == "granted":
print("Notification permission granted.")
else:
print("Notification permission not granted.")
def retrieve_and_save_device_token(self):
from anvil.js import await_promise
token_options = {}
if self.public_vapid_key:
token_options["vapidKey"] = self.public_vapid_key
try:
token = await_promise(self.messaging.getToken(token_options))
if token:
self.device_token = token
anvil.server.call_s("save_device_token", self.device_token)
print(f"Device Token: {self.device_token}")
else:
print("No device token available.")
except Exception as e:
print(f"Error retrieving the device token: {e}")
def setup_token_refresh_handler(self):
self.messaging.onTokenRefresh(self.retrieve_and_save_device_token)
def setup_message_handler(self):
self.messaging.onMessage(self.message_handler)
@anvil.server.portable_class
class PushNotificationMessage:
def __init__(
self,
title: str,
body: str,
icon: str = None,
url_to_open_on_click: str = None,
image: str = None,
):
self.title = title
self.body = body
self.icon = icon
self.url_to_open_on_click = url_to_open_on_click
self.image = image
@property
def as_dict(self):
d = {
"title": self.title,
"body": self.body,
}
if self.icon:
d["icon"] = self.icon
if self.url_to_open_on_click:
d["click_action"] = self.url_to_open_on_click
if self.image:
d["image"] = self.image
return d
def __repr__(self):
return f"<{str(self)}>"
def __str__(self):
return f"PushNotificationMessage(title={self.title}, body={self.body}, icon={self.icon}, url_to_open_on_click={self.url_to_open_on_click}, image={self.image})"
service-worker.js CODE
// Import the Firebase scripts
importScripts('https://www.gstatic.com/firebasejs/7.20.0/firebase-app.js');
importScripts('https://www.gstatic.com/firebasejs/7.20.0/firebase-messaging.js');
// Initialize the Firebase app in the service worker by passing the project configuration
firebase.initializeApp({
apiKey: "",
authDomain: "",
projectId: "",
storageBucket: "",
messagingSenderId: "",
appId: "",
measurementId: ""
}); // I have removed the config date, but it is in my code
// Retrieve an instance of Firebase Messaging
const messaging = firebase.messaging();
// If you would like to customize notifications that are received in the background
messaging.setBackgroundMessageHandler(function (payload) {
console.log('[firebase-messaging-sw.js] Received background message ', payload);
// Customize notification here
const notificationTitle = 'Background Message Title';
const notificationOptions = {
body: 'Background Message body.',
icon: '/firebase-logo.png'
};
return self.registration.showNotification(notificationTitle,
notificationOptions);
});
self.addEventListener('push', function (event) {
const payload = event.data ? event.data.text() : 'no payload';
event.waitUntil(
self.registration.showNotification('Notification Title', {
body: payload,
icon: 'url_to_icon.png',
})
);
});
FIREBASE SERVER CODE
import json
import os
import tempfile
import uuid
import anvil.http
import anvil.media
import anvil.secrets
import anvil.server
import anvil.tables as tables
import anvil.tables.query as q
import anvil.users
import firebase_admin
from anvil.tables import app_tables
from firebase_admin import credentials, initialize_app, messaging
from .FirebaseClient import PushNotificationMessage
def send_notification(
push_notification_message: PushNotificationMessage,
device_token: str,
service_account_path: str,
sound: str = "default",
badge: int = None,
data: dict = None,
):
"""
Sends a push notification to a specified device using Firebase Cloud Messaging (FCM).
This function sends a notification based on the provided `PushNotificationMessage` instance,
which includes options like title, body, icon, URL to open on click, and an image.
Additional features like sound, badge count, and custom data can also be specified.
Parameters:
- push_notification_message (PushNotificationMessage): An instance of PushNotificationMessage
containing the details of the notification.
- device_token (str): The FCM device token of the recipient device.
- access_token (str): OAuth2.0 access token for FCM authentication.
- project_id (str): Firebase project ID.
- sound (str, optional): Notification sound. Default is "default". For custom sounds, provide
the filename of the sound resource.
- badge (int, optional): Badge count to be displayed on the app icon on iOS devices. If None, no badge is set.
- data (dict, optional): Additional custom data to be sent with the notification. This should be a dictionary
where both keys and values are strings.
Returns:
None: The function prints the success message with the response or error message upon failure.
Example:
>>> message = PushNotificationMessage(title="Hello", body="World", icon="icon.png", url_to_open_on_click="https://example.com", image="image.png")
>>> send_notification(message, "<device_token>", "<access_token>", "<project_id>", sound="my_sound.mp3", badge=1, data={"key1": "value1", "key2": "value2"})
Note:
- Ensure that the `access_token` has the required permissions to use FCM services.
- The `device_token` should be a valid token registered with FCM for the target device.
- The `project_id` must correspond to the Firebase project that the notification is being sent from.
- For the `data` dictionary, ensure that the data size does not exceed FCM limits.
"""
# Initialize Firebase Admin
cred = credentials.Certificate(service_account_path)
if not firebase_admin._apps:
initialize_app(cred)
webpush_notification = messaging.WebpushNotification(
title=push_notification_message.title,
body=push_notification_message.body,
icon=push_notification_message.icon,
renotify=True,
)
webpush_notification = messaging.WebpushNotification(
title='Hello, World!',
body='This is a WebPush message from Firebase',
icon='https://example.com/icon.png'
)
webpush_config = messaging.WebpushConfig(
notification=webpush_notification,
)
message = messaging.Message(
webpush=webpush_config,
token=device_token,
)
# Send the notification
try:
response = messaging.send(message)
print(f"Notification sent successfully. Response: {response}")
return response
except Exception as e:
print(f"Error sending notification: {e}")
@anvil.server.callable
def send_test_notification(
push_notification_message: PushNotificationMessage,
device_token: str,
sound: str = "default",
badge: int = None,
data: dict = None,
):
# Get the path to the service account credentials
FCM_SERVICE_ACCOUNT_JSON = anvil.secrets.get_secret("FCM_SERVICE_ACCOUNT_JSON")
# Create a temporary path to the service account credentials
service_account_path = create_temp_path_from_json_string(FCM_SERVICE_ACCOUNT_JSON)
# Send the notification
send_notification(
push_notification_message,
device_token,
service_account_path,
sound,
badge,
data,
)
@anvil.server.http_endpoint("/send_notification", methods=["POST"])
def send_notification_http_endpoint():
print("send_notification_http_endpoint")
# Extract service account path and request body
FCM_SERVICE_ACCOUNT_JSON = anvil.secrets.get_secret("FCM_SERVICE_ACCOUNT_JSON")
service_account_path = create_temp_path_from_json_string(FCM_SERVICE_ACCOUNT_JSON)
request_body = anvil.server.request.body_json
# Extract notification details
device_token = request_body.get("device_token")
push_notification_message = PushNotificationMessage(
title=request_body.get("title"),
body=request_body.get("body"),
image=request_body.get("image"),
url_to_open_on_click=request_body.get("url_to_open_on_click"),
)
print(f"Message: {push_notification_message}")
# Send the notification and handle response
try:
response = send_notification(
push_notification_message,
device_token,
service_account_path,
sound=request_body.get("sound", "default"),
badge=request_body.get("badge"),
data=request_body.get("data"),
)
print(f"Notification sent successfully. Response: {response}")
return anvil.server.HttpResponse(status=200)
except Exception as e:
print(f"Error sending notification: {e}")
return anvil.server.HttpResponse(status=500)
def create_temp_path_from_json_string(json_string):
try:
# Create a temporary file
temp_file = tempfile.NamedTemporaryFile(delete=False, mode="w", suffix=".json")
temp_file.write(json_string)
temp_file.close()
return temp_file.name
except json.JSONDecodeError as e:
raise ValueError("Invalid JSON string") from e
except Exception as ex:
raise Exception("Error creating temporary file") from ex
def get_or_create_token_row(mac_address):
# Try to get the row in the tokens table that matches the MAC address
token_row = app_tables.tokens.get(mac_address=mac_address)
# If no row is found, create a new one with the MAC address and IP address
if not token_row:
ip_address = anvil.server.context.client.ip
token_row = app_tables.tokens.add_row(
mac_address=mac_address, ip_address=ip_address
)
return token_row
@anvil.server.callable
def get_device_token():
# Get the MAC address of the current device
mac_address = uuid.getnode()
# Get the token row, creating one if necessary
token_row = get_or_create_token_row(mac_address)
# Get the list of tokens for the row, or an empty list if none exist
tokens = token_row["tokens"] or []
# If there is at least one token, return the first token
if len(tokens) > 0:
return tokens[0]
@anvil.server.callable
def save_device_token(token):
# Get the MAC address of the current device
mac_address = uuid.getnode()
# Get the token row, creating one if necessary
token_row = get_or_create_token_row(mac_address)
# Get the list of tokens for the row, or an empty list if none exist
tokens = token_row["tokens"] or []
# Add the new token to the list
tokens.append(token)
# Update the row with the new list of tokens
token_row["tokens"] = tokens
@anvil.server.callable
def get_ip():
return anvil.server.context.client.ip
When I try run the app in the designer it seems to load up correctly and returns a device_token
. I can then trigger the send_test_notification
function on the server and that returns a “Successful” execution, but NO NOTIFICATION ever comes through. I have tried doing this from my phone with the app URL, and I get nothing there either. I have even tried triggering the send_notification
function via an HTTP call from Postman, and nothing seems to work. Also, when I try running this app on Brave, I get all sorts of errors in the console: Manifest Icon Error, Service Worker Fetch Error, and PostMessage Error. I’m at a complete loss right now.
ANY HELP IS GREATLY APPRECIATED!