Modul

pullkin.core.PullkinCore

Core methods for GCM and FCM registration.

This class contains all the underlying methods for GCM and FCM registration. However, users should usually only care about the Pullkin.register() method.

fcm_installation(app_id, api_key, android_cert, app_package, firebase_project_id) async

Performs FCM installation.

Parameters:

Name	Type	Description	Default
app_id	str	App ID of the Firebase app.	required
api_key	str	API key of the Firebase app.	required
android_cert	str	Android certificate of the Firebase app.	required
app_package	str	Package name of the app.	required
firebase_project_id	str	Firebase project ID of the app.	required

Returns:

Type	Description
FirebaseInstallation	The device installation data of the firebase app.

fcm_register(gcm_token, api_key, firebase_project_id, firebase_installation, retries=5) async

Generates a key pair and registers a GCM token.

This method may return an error in the AppCredentialsFcm.error field. To validate the success of the registration, check the AppCredentialsFcm.is_success field. For more details about the error, if any, check the AppCredentialsFcm.error field.

Parameters:

Name	Type	Description	Default
gcm_token	str	The subscription GCM token returned by Pullkin.gcm_register(...).	required
api_key	str	The API key for the Firebase project.	required
firebase_project_id	str	The ID of the Firebase project.	required
firebase_installation	FirebaseInstallation	The Firebase installation data.	required
retries	int	The number of retries for the registration.	5

Returns:

Type	Description
tuple[AppCredentialsKeys, AppCredentialsFcm]	A tuple containing the generated key pair and the FCM registration data.

gcm_check_in(credentials=None) async

Perform a check-in request to obtain the android_id and security token.

The android_id and security_token are required for the GCM registration, and will be returned in the response.

Parameters:

Name	Type	Description	Default
credentials	Optional[AppCredentialsGcm]	The credentials to use for the GCM check-in. Required for first check-in.	None

Returns:

Type	Description
AndroidCheckinResponse	The response to the check-in request, containing the android_id, security_token and more.

gcm_register(sender_id, app_id, api_key=None, firebase_project_id=None, android_cert=None, app_name='org.chromium.linux', retries=5, current_retry=0, **_) async

Registers a device to GCM.

This method performs the GCM device registration, which is a mandatory step before a device can receive push notifications. The method performs the following steps:

1. Performs a Chrome build check-in to obtain the Android ID and security token.
2. Registers the device to GCM using the obtained Android ID and security token.

The registration to Firebase is optional and can be skipped if the required parameters are not provided.

Parameters:

Name	Type	Description	Default
sender_id	str	The Sender ID of the app.	required
app_id	str	The app ID in the form of "1:123123:android".	required
api_key	str	The API key in the form of "AIzaSy...".	None
firebase_project_id	str	The project ID of the Firebase project.	None
android_cert	str	The android cert hash as a base64 string.	None
app_name	str	The package name of the app.	'org.chromium.linux'
retries	int	The number of failed requests before giving up.	5
current_retry	int	The current retry counts. Default to 0.	0

Returns:

Type	Description
AppCredentialsGcm	The GCM registration data.

register(sender_id, app_id, api_key, firebase_project_id, android_cert='da39a3ee5e6b4b0d3255bfef95601890afd80709', app_name='org.chromium.linux', retries=5) async

Registers a device in GCM and FCM.

Firebase installation is optional and can be skipped if parameters are not provided.

FCM subscription may contain AppCredentials.fcm.error if not compatible.

Parameters:

Name	Type	Description	Default
sender_id	int | str	sender_id, required	required
app_id	str	app_id in the form of "1:123123:android"	required
api_key	str	api_key in the form of "AIzaSy..."	required
firebase_project_id	str	firebase project name	required
android_cert	str	android cert hash as a base64 string	'da39a3ee5e6b4b0d3255bfef95601890afd80709'
app_name	str	package name, "org.chromium.linux" by default	'org.chromium.linux'
retries	int	number of failed requests before giving up	5

Returns:

Type	Description
AppCredentials	GCM and FCM data and keys.
AppCredentials	If fcm registration fails,AppCredentialsGcm.fcm.error
AppCredentials	will contain a PullkinResponseError with more details.

pullkin.Pullkin

Bases: PullkinCore

on_notification(handler_filter=lambda *a, **k: True)

Registers a new callback with a filter rule to trigger this callback.

Parameters:

Name	Type	Description	Default
handler_filter	Callable[[Message, DataMessageStanza], None]	A callable object that filters notifications. It takes two parameters: the message and the data message stanza. If the callable returns True, the callback will be triggered.	lambda *a, **k: True

Returns:

Type	Description
Callable[[T], T]	A decorator that registers the callback.

Example

from pullkin import Pullkin, Message, DataMessageStanza
pullkin = AuiPullkin()

@pullkin.on_notification(lambda m, d: m.notification.title == "Moi koleni zamerzli")
def on_zemfira(message: Message, data_message: DataMessageStanza):
    print(message.notification.body)

@pullkin.on_notification(lambda m, d: m.priority == "normal")
async def on_normal(message: Message, data_message: DataMessageStanza):
    print(f"#{data_message.id} @{data_message.sent}")
    print(message.notification)
    await asyncio.sleep(2)
    print("=-)")

listen_coroutine(sender_id=None) async

Creates a coroutine to listen for push notifications.

This coroutine yields a received push notification on each iteration. If no notification is available, it waits for data on the socket.

Parameters:

Name	Type	Description	Default
sender_id	int | str | None	The sender ID to filter notifications for. If None, listen to all notifications.	None

Returns:

Type	Description
AsyncGenerator[Message | None, None]	An asynchronous generator that yields each received push notification, or None if no notification is received.

Yields:

Type	Description
AsyncGenerator[Message | None, None]	Message | None: A received push notification message, or None if no notification is received.

Example

# Perform some initial setup...

# Receive and distribute one push notification.
await coroutine.asend(None)

# Perform some other tasks...

# Receive and distribute ten push notifications.
for _ in range(10):
    await coroutine.asend(None)

# Perform some final tasks...

add_app(sender_id, credentials, persistent_ids=None) async

Subscribe a device to a specific app.

This method is used to subscribe a device to a specific app, allowing it to receive notifications from that app.

Parameters:

Name	Type	Description	Default
sender_id	str	The sender ID of the app.	required
credentials	AppCredentials	The credentials for the app.	required
persistent_ids	set[str]	The persistent IDs for the app.	None

Returns:

Type	Description
None	None

run(sender_ids=None, timer=0.05) async

Listens for push notifications in the background.

This method runs an endless loop that reads incoming notifications and distributes them among registered callbacks based on filter results.

Parameters:

Name	Type	Description	Default
sender_ids	list[str] | None	A list of sender IDs to listen for notifications from. If None, all registered via add_app() sender IDs will be listened to.	None
timer	int | float	The time interval in seconds between each receive-iteration. Defaults to 0.05 seconds.	0.05

Returns:

Type	Description
None	None

stop(sender_id=None) async

Closes the connection for the specified sender ID or all sender IDs if None is provided.

Parameters:

Name	Type	Description	Default
sender_id	str | None	The sender ID to close the connection for. None for all.	None

Returns:

Type	Description
None	None