APNs Integration Guide¶

from homepot.push_notifications import get_push_provider
from homepot.push_notifications.base import (
    PushNotificationPayload,
    PushPriority
)

# Configuration
apns_config = {
    "team_id": "ABC123DEFG",
    "key_id": "XYZ987WXYZ",
    "auth_key_path": "/path/to/AuthKey_XYZ987WXYZ.p8",
    "bundle_id": "com.homepot.client",
    "environment": "production",
    "topic": "com.homepot.client"
}

# Initialize APNs provider
provider = await get_push_provider("apns", apns_config)

# Device token (from your iOS/macOS app)
device_token = "a1b2c3d4e5f6789012345678901234567890abcdefabcdef1234567890abcd"

# Create notification
payload = PushNotificationPayload(
    title="Payment Pending",
    body="Transaction #12345 awaiting approval at Terminal POS-001",
    data={
        "transaction_id": "12345",
        "pos_id": "POS-001",
        "amount": "150.00",
        "action": "approve_payment"
    },
    priority=PushPriority.HIGH
)

# Send notification
result = await provider.send_notification(device_token, payload)

if result.success:
    print(f"Notification sent! ID: {result.message_id}")
else:
    print(f"Failed: {result.message} (Error: {result.error_code})")

payload = PushNotificationPayload(
    title="New Transaction",
    body="Customer payment received: $150.00",
    platform_data={
        "badge": 5,          # Red badge number on app icon
        "sound": "default"   # Plays notification sound
    }
)

payload = PushNotificationPayload(
    title="Background Update",
    body="Syncing configuration",
    platform_data={
        "content_available": True,  # Enable background mode
        "sound": None                # No sound
    }
)

payload = PushNotificationPayload(
    title="Badge Update",
    body="",
    platform_data={
        "badge": 3,      # Update to 3 pending items
        "sound": None    # Silent
    }
)

payload = PushNotificationPayload(
    title="High Priority Alert",
    body="System error detected",
    platform_data={
        "sound": "critical_alert.caf"  # Custom sound file
    }
)

payload = PushNotificationPayload(
    title="Transaction Pending",
    body="Approve or decline transaction #12345?",
    platform_data={
        "category": "TRANSACTION_ACTIONS"  # Matches category in your app
    }
)

# High priority (immediate)
payload = PushNotificationPayload(
    title="Critical Alert",
    body="Payment gateway offline",
    priority=PushPriority.HIGH  # Delivers immediately
)

# Normal priority (battery-efficient)
payload = PushNotificationPayload(
    title="Daily Report",
    body="Sales summary ready",
    priority=PushPriority.NORMAL  # May delay delivery
)

# List of (device_token, payload) tuples
notifications = [
    ("device_token_1", payload_for_device_1),
    ("device_token_2", payload_for_device_2),
    ("device_token_3", payload_for_device_3),
    # ... up to hundreds of devices
]

# Send all at once
results = await provider.send_bulk_notifications(notifications)

# Check results
for result in results:
    if result.success:
        print(f"Sent to {result.device_token}")
    else:
        print(f"Failed {result.device_token}: {result.error_code}")

result = await provider.send_notification(device_token, payload)

if result.error_code == "UNREGISTERED":
    # User uninstalled the app - remove token from database
    await database.delete_device_token(device_token)
    print(f"Removed unregistered device: {device_token}")

result = await provider.send_notification(device_token, payload)

if result.success:
    print(f"Notification sent")

elif result.error_code == "UNREGISTERED":
    # App uninstalled - remove from database
    await remove_device_from_database(device_token)

elif result.error_code == "AUTH_FAILED":
    # Credentials problem
    print("Check your Team ID, Key ID, and .p8 file")

elif result.error_code == "TOO_MANY_REQUESTS":
    # Rate limited - wait and retry
    await asyncio.sleep(result.retry_after or 60)
    result = await provider.send_notification(device_token, payload)

elif result.error_code == "SERVER_ERROR":
    # APNs server issue - retry
    await asyncio.sleep(30)
    result = await provider.send_notification(device_token, payload)

else:
    print(f"Error: {result.message}")

apns_config = {
    # ... other config
    "environment": "sandbox"
}

apns_config = {
    # ... other config
    "environment": "production"
}

import json

# Build payload
payload = PushNotificationPayload(
    title="Very Long Title" * 100,  # Too long!
    body="Message",
    data={"large_data": "x" * 5000}  # Too much data!
)

# APNs provider will automatically check
result = await provider.send_notification(device_token, payload)

if result.error_code == "PAYLOAD_TOO_LARGE":
    print(f"Payload exceeds 4KB limit - reduce content")

# Check if APNs is working
provider = await get_push_provider("apns", apns_config)
health = await provider.health_check()

print(f"Status: {health['status']}")
print(f"Environment: {health['platform_info']['environment']}")
print(f"JWT Valid: {health['platform_info']['jwt_valid']}")

if result.error_code == "UNREGISTERED":
    # Delete from database
    await db.execute(
        "DELETE FROM device_tokens WHERE token = ?",
        (device_token,)
    )

# Validate token format
token = device_token.strip()  # Remove whitespace
if len(token) != 64:
    print(f"Invalid token length: {len(token)} (should be 64)")

# Before (too large)
payload = PushNotificationPayload(
    title="Transaction Report",
    body="Here are all the details...",
    data={
        "full_report": "... 5KB of data ..."  # Too much!
    }
)

# After (optimized)
payload = PushNotificationPayload(
    title="Report Ready",
    body="Tap to view",
    data={
        "report_id": "12345"  # Just the ID, fetch details in app
    }
)

# Use bulk send for many devices
results = await provider.send_bulk_notifications(notifications)

# Or add delays between sends
for device_token, payload in notifications:
    result = await provider.send_notification(device_token, payload)

    if result.error_code == "TOO_MANY_REQUESTS":
        # Wait before retrying
        wait_time = result.retry_after or 60
        await asyncio.sleep(wait_time)
        result = await provider.send_notification(device_token, payload)

import asyncio
from homepot.push_notifications import get_push_provider
from homepot.push_notifications.base import (
    PushNotificationPayload,
    PushPriority
)

class POSNotificationSystem:
    def __init__(self):
        self.apns_config = {
            "team_id": "ABC123DEFG",
            "key_id": "XYZ987WXYZ",
            "auth_key_path": "/secure/AuthKey_XYZ987WXYZ.p8",
            "bundle_id": "com.homepot.pos",
            "environment": "production",
            "topic": "com.homepot.pos"
        }
        self.provider = None

    async def initialize(self):
        """Initialize APNs provider."""
        self.provider = await get_push_provider("apns", self.apns_config)
        health = await self.provider.health_check()

        if health['status'] != 'healthy':
            raise RuntimeError("APNs provider not healthy")

        print("APNs initialized successfully")

    async def notify_payment_pending(
        self,
        device_token: str,
        transaction_id: str,
        amount: float,
        terminal_id: str
    ):
        """Send payment approval notification."""

        payload = PushNotificationPayload(
            title="Payment Approval Required",
            body=f"Transaction #{transaction_id}: ${amount:.2f}",
            data={
                "transaction_id": transaction_id,
                "amount": str(amount),
                "terminal_id": terminal_id,
                "action": "approve_payment",
                "timestamp": datetime.utcnow().isoformat()
            },
            priority=PushPriority.HIGH,
            platform_data={
                "badge": 1,
                "sound": "payment_alert.caf",
                "category": "PAYMENT_APPROVAL"
            }
        )

        result = await self.provider.send_notification(
            device_token,
            payload
        )

        if result.success:
            print(f"Payment notification sent: {result.message_id}")
            return True
        elif result.error_code == "UNREGISTERED":
            # Device uninstalled app - remove from database
            await self.remove_device(device_token)
            print(f"Removed unregistered device")
            return False
        else:
            print(f"Notification failed: {result.message}")
            return False

    async def notify_configuration_update(self, device_tokens: list):
        """Send silent background update to multiple devices."""

        payload = PushNotificationPayload(
            title="Configuration Update",
            body="",
            data={
                "update_type": "configuration",
                "version": "2.1.0"
            },
            priority=PushPriority.NORMAL,
            platform_data={
                "content_available": True,  # Silent notification
                "sound": None
            }
        )

        notifications = [
            (token, payload) for token in device_tokens
        ]

        results = await self.provider.send_bulk_notifications(notifications)

        success_count = sum(1 for r in results if r.success)
        print(f"Configuration update sent to {success_count}/{len(results)} devices")

        # Handle unregistered devices
        for result in results:
            if result.error_code == "UNREGISTERED":
                await self.remove_device(result.device_token)

    async def remove_device(self, device_token: str):
        """Remove device from database."""
        # Your database logic here
        print(f"Removing device: {device_token}")

# Usage
async def main():
    system = POSNotificationSystem()
    await system.initialize()

    # Send payment notification
    await system.notify_payment_pending(
        device_token="a1b2c3d4e5f6...64chars...",
        transaction_id="TXN-12345",
        amount=150.00,
        terminal_id="POS-001"
    )

    # Send config update to all devices
    all_devices = [
        "device_token_1",
        "device_token_2",
        "device_token_3"
    ]
    await system.notify_configuration_update(all_devices)

if __name__ == "__main__":
    asyncio.run(main())