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