Async Firebase Client - a Python asyncio client to interact with Firebase Cloud Messaging in an easy way.
firebase-admin which is heftyTo install async-firebase, simply execute the following command in a terminal:
$ pip install async-firebase
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())
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())
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())
async-firebase is offered under the MIT license.
The latest developer version is available in a GitHub repository: https://github.com/healthjoy/async-firebase
FAQs
Async Firebase Client - a Python asyncio client to interact with Firebase Cloud Messaging in an easy way.
We found that async-firebase demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
PyPI confirms no security flaws were exploited in the Ultralytics supply chain attack and highlights improvements for safer package publishing.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.