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.
Research
Security News
Socket’s threat research team has detected six malicious npm packages typosquatting popular libraries to insert SSH backdoors.
Security News
MITRE's 2024 CWE Top 25 highlights critical software vulnerabilities like XSS, SQL Injection, and CSRF, reflecting shifts due to a refined ranking methodology.
Security News
In this segment of the Risky Business podcast, Feross Aboukhadijeh and Patrick Gray discuss the challenges of tracking malware discovered in open source softare.