Skip to main content

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

Project description

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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

async_firebase-3.8.0.tar.gz (24.5 kB view hashes)

Uploaded Source

Built Distribution

async_firebase-3.8.0-py3-none-any.whl (26.5 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page