Firebase FCM for Push Notifications

I was working on a project for one of my clients, and the team received a requirement to implement Push Notifications across Android, iOS, and Web Portals. So here’s a step-by-step guide on how to achieve this using Firebase Cloud Messaging (FCM).

Before you start coding, make sure your project is correctly set up for Firebase Cloud Messaging (FCM) in the Firebase Console. For push notifications, you’ll need a service-key.json file from Firebase, which contains your service account credentials. You can download this file from the Firebase Project Settings under the Service Accounts tab.

Code :

import asyncio
import firebase_admin
from firebase_admin import messaging

# Initialize Firebase Admin SDK
try:
    cred = credentials.Certificate("./service-key.json")
    firebase_admin.initialize_app(cred)
except Exception as e:
    print(f"Error initializing Firebase Admin SDK: {e}")

def extract_platform_and_token(token_with_metadata):
    """
    Extracts the platform and FCM token from the stored metadata.
    :param token_with_metadata: The stored token with metadata in format 'platform|token'
    :return: A tuple (platform, fcm_token)
    """
    try:
        platform, fcm_token = token_with_metadata.split('|', 1)
        return platform, fcm_token
    except ValueError:
        return None, None

async def send_notification(data, deviceTokenInput):
    def _send_notification():
        try:
            platform, device_token = extract_platform_and_token(deviceTokenInput)
            title = data.get("title_html", "Default Title")
            body = data.get("message", "Default Message")
            download_url = data.get("download_url", "")

            if platform == "ios":
                apns_config = messaging.APNSConfig(
                    payload=messaging.APNSPayload(
                        aps=messaging.Aps(
                            alert=messaging.ApsAlert(
                                title=title,
                                body=body
                            ),
                            sound="default",
                            badge=1
                        )
                    ),
                    fcm_options=messaging.APNSFCMOptions(
                        image=download_url
                    )
                )
                message = messaging.Message(apns=apns_config, token=device_token)

            elif platform == "android":
                android_config = messaging.AndroidConfig(
                    notification=messaging.AndroidNotification(
                        title=title,
                        body=body,
                        sound="default",
                        image=download_url
                    ),
                    priority="high"
                )
                message = messaging.Message(android=android_config, token=device_token)

            elif platform == "web":
                webpush_config = messaging.WebpushConfig(
                    headers={"Urgency": "high"},
                    notification=messaging.WebpushNotification(
                        title=title,
                        body=body,
                        icon=download_url
                    ),
                    fcm_options=messaging.WebpushFCMOptions(link=download_url)
                )
                message = messaging.Message(webpush=webpush_config, token=device_token)

            else:
                raise ValueError("Invalid platform. Choose from 'web', 'android', or 'ios'.")

            messaging.send(message)

        except Exception as e:
            print(f"Error sending message: {e}")

    return await asyncio.to_thread(_send_notification)

async def main():
    data = {
        "id": "1",
        "type": "notification",
        "title_html": "SQA Test Push",
        "read": False,
        "time": "2024-08-08T10:30:00",
        "message": "Your tour has been booked successfully!",
        "flagged": False,
        "tour_id": "100",
        "download_url": "https://example.com/image.png"
    }

    ios_token = "ios|<fcm token generated for ios device>"
    web_token = "web|<fcm token generated for web>"
    android_token = "android|<fcm token generated for andoid device>"

    await send_notification(data, web_token)
    await send_notification(data, ios_token)
    await send_notification(data, android_token)

asyncio.run(main())