async-firebase 5.0.0

Async Firebase Client - a Python asyncio client to interact with Firebase Cloud Messaging in an easy way.

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

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

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