Web Push Notifications to Clients

What I’m trying to do:
How can I implement Web Push Notifications? I want to create an app that can alert a user when it’s their turn to do something.

What I’ve tried and what’s not working:
Nothing yet

Code Sample:

# code snippet

Clone link:
share a copy of your app

“Web push notification” is a particular sort of notification that’ll show up on the user’s device even if the app isn’t running. Those typically involves a third-party service of some sort. There’d be some Javascript of theirs to include in your native libraries section so that users can opt-in to (or out of) notifications. Then there’d be a call from your server code to the third-party API to send the notification.

All that is doable in Anvil, but you’d need to work your way through all the steps after picking a third-party service. sendpulse.com has a generous free tier you can use to play with it.

I’d suggest signing up for a free account at sendpulse or another service, create a simple Anvil app to test notifications with, and start working through the process. When you hit a roadblock, ask here but provide a clone link to your test app so we can see what you’re doing.

Alternatively, if you only want notifications to show when the user has the app running, then a Timer component on the form will allow you to do periodic server calls to see if the user has any notifications that need displayed.

1 Like

Thanks, I’ll check out sendpulse.com

A while back, I had some luck using this basic JavaScript API for notifications that show up while the app is running:

1 Like

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. :rage: 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!

Have you tried this

I did look through this post, but unless I missed something, it appeared to using the legacy API which is getting ready to be obsolete in a few days.

For that reason I decided it would be better for me to slam my head against the wall for past few days :crazy_face:

I “think” i’ve solved the issue. I just need to do a little more testing and code refinement and I plan no posting my solution.

Glad that you solved it. Also, you are correct that the dependency hasn’t been upgraded yet (Although I still have 6+ months to update the dependency and I will be doing that soon :slightly_smiling_face:)

Hey everyone,

A while back, I reached out to this amazing community for guidance on integrating Firebase with Anvil. Today, I’m thrilled to announce that, thanks to your invaluable advice and support, I’ve successfully completed the Firebase integration for Anvil apps!

I’m excited to share this accomplishment with all of you who have helped and inspired me along this journey. To showcase what I’ve built and to give back to the community that has given me so much, I’ve created a detailed Show and Tell post. It includes a thorough guide, and examples.

You can find my post here: New Anvil Firebase Integration - Show and Tell - Anvil Community Forum. I hope it serves as a helpful resource for anyone looking to implement Firebase in their Anvil projects.

Once again, thank you all for your support and guidance. This community is truly a goldmine of knowledge and camaraderie, and I’m proud to be a part of it.