Skip to content

The lightweight asynchronous client that makes interaction with Firebase Cloud Messaging oversimplified.

License

Notifications You must be signed in to change notification settings

healthjoy/async-firebase

Repository files navigation

async-firebase is a lightweight asynchronous client to interact with Firebase Cloud Messaging for sending push notification to Android and iOS devices

PyPI download month PyPI version fury.io PyPI license PyPI pyversions CI Codacy coverage

  • Free software: MIT license
  • Requires: Python 3.8+

Features

  • Extremely lightweight and does not rely on firebase-admin which is hefty
  • Send push notifications to Android and iOS devices
  • Send Multicast push notification to Android and iOS devices
  • Send Web push notifications
  • Set TTL (time to live) for notifications
  • Set priority for notifications
  • Set collapse-key for notifications
  • Dry-run mode for testing purpose
  • Topic Management

Installation

To install async-firebase, simply execute the following command in a terminal:

$ pip install async-firebase

Getting started

async-firebase < 3.0.0

To send push notification to Android:

import asyncio

from async_firebase import AsyncFirebaseClient


async def main():
    client = AsyncFirebaseClient()
    client.creds_from_service_account_file("secret-store/mobile-app-79225efac4bb.json")

    # or using dictionary object
    # client.creds_from_service_account_info({...}})

    device_token = "..."

    android_config = client.build_android_config(
        priority="high",
        ttl=2419200,
        collapse_key="push",
        data={"discount": "15%", "key_1": "value_1", "timestamp": "2021-02-24T12:00:15"},
        title="Store Changes",
        body="Recent store changes",
    )
    response = await client.push(device_token=device_token, android=android_config)

    print(response.success, response.message_id)

if __name__ == "__main__":
    asyncio.run(main())

To send push notification to iOS:

import asyncio

from async_firebase import AsyncFirebaseClient


async def main():
    client = AsyncFirebaseClient()
    client.creds_from_service_account_file("secret-store/mobile-app-79225efac4bb.json")

    # or using dictionary object
    # client.creds_from_service_account_info({...}})

    device_token = "..."

    apns_config = client.build_apns_config(
        priority="normal",
        ttl=2419200,
        apns_topic="store-updated",
        collapse_key="push",
        title="Store Changes",
        alert="Recent store changes",
        badge=1,
        category="test-category",
        custom_data={"discount": "15%", "key_1": "value_1", "timestamp": "2021-02-24T12:00:15"}
    )
    response = await client.push(device_token=device_token, apns=apns_config)

    print(response.success)

if __name__ == "__main__":
    asyncio.run(main())

This prints:

"projects/mobile-app/messages/0:2367799010922733%7606eb557606ebff"

To manual construct message:

import asyncio
from datetime import datetime

from async_firebase.messages import APNSConfig, APNSPayload, ApsAlert, Aps
from async_firebase import AsyncFirebaseClient


async def main():
    apns_config = APNSConfig(**{
        "headers": {
            "apns-expiration": str(int(datetime.utcnow().timestamp()) + 7200),
            "apns-priority": "10",
            "apns-topic": "test-topic",
            "apns-collapse-id": "something",
        },
        "payload": APNSPayload(**{
            "aps": Aps(**{
                "alert": ApsAlert(title="some-title", body="alert-message"),
                "badge": 0,
                "sound": "default",
                "content_available": True,
                "category": "some-category",
                "mutable_content": False,
                "custom_data": {
                    "link": "https://link-to-somewhere.com",
                    "ticket_id": "YXZ-655512",
                },
            })
        })
    })

    device_token = "..."

    client = AsyncFirebaseClient()
    client.creds_from_service_account_info({...})
    response = await client.push(device_token=device_token, apns=apns_config)
    print(response.success)


if __name__ == "__main__":
    asyncio.run(main())

async-firebase >= 3.0.0

To send push notification to Android:

import asyncio

from async_firebase import AsyncFirebaseClient
from async_firebase.messages import Message


async def main():
    client = AsyncFirebaseClient()
    client.creds_from_service_account_file("secret-store/mobile-app-79225efac4bb.json")

    # or using dictionary object
    # client.creds_from_service_account_info({...}})

    device_token: str = "..."

    android_config = client.build_android_config(
        priority="high",
        ttl=2419200,
        collapse_key="push",
        data={"discount": "15%", "key_1": "value_1", "timestamp": "2021-02-24T12:00:15"},
        title="Store Changes",
        body="Recent store changes",
    )
    message = Message(android=android_config, token=device_token)
    response = await client.send(message)

    print(response.success, response.message_id)

if __name__ == "__main__":
    asyncio.run(main())

To send push notification to iOS:

import asyncio

from async_firebase import AsyncFirebaseClient
from async_firebase.messages import Message


async def main():
    client = AsyncFirebaseClient()
    client.creds_from_service_account_file("secret-store/mobile-app-79225efac4bb.json")

    # or using dictionary object
    # client.creds_from_service_account_info({...}})

    device_token: str = "..."

    apns_config = client.build_apns_config(
        priority="normal",
        ttl=2419200,
        apns_topic="store-updated",
        collapse_key="push",
        title="Store Changes",
        alert="Recent store changes",
        badge=1,
        category="test-category",
        custom_data={"discount": "15%", "key_1": "value_1", "timestamp": "2021-02-24T12:00:15"}
    )
    message = Message(apns=apns_config, token=device_token)
    response = await client.send(message)

    print(response.success)

if __name__ == "__main__":
    asyncio.run(main())

This prints:

"projects/mobile-app/messages/0:2367799010922733%7606eb557606ebff"

To manual construct message:

import asyncio
from datetime import datetime

from async_firebase.messages import APNSConfig, APNSPayload, ApsAlert, Aps, Message
from async_firebase import AsyncFirebaseClient


async def main():
    apns_config = APNSConfig(**{
        "headers": {
            "apns-expiration": str(int(datetime.utcnow().timestamp()) + 7200),
            "apns-priority": "10",
            "apns-topic": "test-topic",
            "apns-collapse-id": "something",
        },
        "payload": APNSPayload(**{
            "aps": Aps(**{
                "alert": ApsAlert(title="some-title", body="alert-message"),
                "badge": 0,
                "sound": "default",
                "content_available": True,
                "category": "some-category",
                "mutable_content": False,
                "custom_data": {
                    "link": "https://link-to-somewhere.com",
                    "ticket_id": "YXZ-655512",
                },
            })
        })
    })

    device_token: str = "..."

    client = AsyncFirebaseClient()
    client.creds_from_service_account_info({...})
    message = Message(apns=apns_config, token=device_token)
    response = await client.send(message)
    print(response.success)


if __name__ == "__main__":
    asyncio.run(main())

Topic Management

You can subscribe and unsubscribe client app instances in bulk approach by passing a list of registration tokens to the subscription method to subscribe the corresponding devices to a topic:

import asyncio

from async_firebase import AsyncFirebaseClient


async def main():
    device_tokens: list[str] = ["...", "..."]

    client = AsyncFirebaseClient()
    client.creds_from_service_account_info({...})
    response = await client.subscribe_devices_to_topic(
        device_tokens=device_tokens, topic_name="some-topic"
    )
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

To unsubscribe devices from a topic by passing registration tokens to the appropriate method:

import asyncio

from async_firebase import AsyncFirebaseClient


async def main():
    device_tokens: list[str] = ["...", "..."]

    client = AsyncFirebaseClient()
    client.creds_from_service_account_info({...})
    response = await client.unsubscribe_devices_from_topic(
        device_tokens=device_tokens, topic_name="some-topic"
    )
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

License

async-firebase is offered under the MIT license.

Source code

The latest developer version is available in a GitHub repository: https://github.com/healthjoy/async-firebase