Add LiveActivity support

kalyke/__init__.py

@@ -1,24 +1,35 @@
 from .clients.apns import ApnsClient
+from .clients.live_activity import LiveActivityClient
 from .clients.voip import VoIPClient
-from .models.apns_config import ApnsConfig, LiveActivityApnsConfig, VoIPApnsConfig
-from .models.apns_priority import ApnsPriority
-from .models.apns_push_type import ApnsPushType
-from .models.critical_sound import CriticalSound
-from .models.interruption_level import InterruptionLevel
-from .models.payload import Payload
-from .models.payload_alert import PayloadAlert
+from .models import (
+    ApnsConfig,
+    ApnsPriority,
+    ApnsPushType,
+    CriticalSound,
+    InterruptionLevel,
+    LiveActivityApnsConfig,
+    LiveActivityEvent,
+    LiveActivityPayload,
+    Payload,
+    PayloadAlert,
+    VoIPApnsConfig,
+)
+
 
 __all__ = [
     "ApnsClient",
-    "VoIPClient",
     "ApnsConfig",
-    "LiveActivityApnsConfig",
-    "VoIPApnsConfig",
     "ApnsPriority",
     "ApnsPushType",
     "CriticalSound",
     "InterruptionLevel",
+    "LiveActivityApnsConfig",
+    "LiveActivityClient",
+    "LiveActivityEvent",
+    "LiveActivityPayload",
     "Payload",
     "PayloadAlert",
+    "VoIPClient",
+    "VoIPApnsConfig",
     "exceptions",
 ]

kalyke/clients/live_activity.py

@@ -0,0 +1,24 @@
+from dataclasses import dataclass
+
+from httpx import AsyncClient
+
+from ..models import LiveActivityApnsConfig, LiveActivityPayload
+from .apns import ApnsClient
+
+
+@dataclass
+class LiveActivityClient(ApnsClient):
+    async def send_message(
+        self,
+        device_token: str,
+        payload: LiveActivityPayload,
+        apns_config: LiveActivityApnsConfig,
+    ) -> str:
+        return await super().send_message(
+            device_token=device_token,
+            payload=payload,
+            apns_config=apns_config,
+        )
+
+    def _init_client(self, apns_config: LiveActivityApnsConfig) -> AsyncClient:
+        return super()._init_client(apns_config=apns_config)

kalyke/clients/voip.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Union
+from typing import Union, Dict, Any
 
 import httpx
 from httpx import AsyncClient
@@ -21,6 +21,18 @@ def __post_init__(self):
         else:
             self._auth_key_filepath = Path(self.auth_key_filepath)
 
+    async def send_message(
+        self,
+        device_token: str,
+        payload: Dict[str, Any],
+        apns_config: VoIPApnsConfig,
+    ) -> str:
+        return await super().send_message(
+            device_token=device_token,
+            payload=payload,
+            apns_config=apns_config,
+        )
+
     def _init_client(self, apns_config: VoIPApnsConfig) -> AsyncClient:
         headers = apns_config.make_headers()
         context = httpx.create_ssl_context()

kalyke/exceptions.py

@@ -21,6 +21,16 @@ def __str__(self) -> str:
         return f"The system uses the relevance_score, a value between 0 and 1. Did set {self._relevance_score}."
 
 
+class LiveActivityAttributesIsNotJSONSerializable(Exception):
+    def __str__(self) -> str:
+        return "The attributes is not JSON serializable."
+
+
+class LiveActivityContentStateIsNotJSONSerializable(Exception):
+    def __str__(self) -> str:
+        return "The content state is not JSON serializable."
+
+
 # https://developer.apple.com/documentation/usernotifications/setting_up_a_remote_notification_server/handling_notification_responses_from_apns#3394535
 class ApnsProviderException(Exception):
     def __init__(self, error: Dict[str, Any]) -> None:

kalyke/models/__init__.py

@@ -3,7 +3,8 @@
 from .apns_push_type import ApnsPushType
 from .critical_sound import CriticalSound
 from .interruption_level import InterruptionLevel
-from .payload import Payload
+from .live_activity_event import LiveActivityEvent
+from .payload import LiveActivityPayload, Payload
 from .payload_alert import PayloadAlert
 
 __all__ = [
@@ -14,6 +15,8 @@
     "ApnsPushType",
     "CriticalSound",
     "InterruptionLevel",
+    "LiveActivityEvent",
     "Payload",
+    "LiveActivityPayload",
     "PayloadAlert",
 ]

kalyke/models/live_activity_event.py

@@ -0,0 +1,7 @@
+from enum import Enum
+
+
+class LiveActivityEvent(Enum):
+    start: str = "start"
+    update: str = "update"
+    end: str = "end"

kalyke/models/payload.py

@@ -1,9 +1,16 @@
+import json
 from dataclasses import dataclass, field
+from datetime import datetime
 from typing import Any, Dict, Optional, Union
 
-from ..exceptions import RelevanceScoreOutOfRangeException
+from ..exceptions import (
+    LiveActivityAttributesIsNotJSONSerializable,
+    LiveActivityContentStateIsNotJSONSerializable,
+    RelevanceScoreOutOfRangeException,
+)
 from .critical_sound import CriticalSound
 from .interruption_level import InterruptionLevel
+from .live_activity_event import LiveActivityEvent
 from .payload_alert import PayloadAlert
 
 
@@ -24,10 +31,13 @@ class Payload:
 
     def __post_init__(self):
         if self.relevance_score:
-            if 0.0 <= self.relevance_score <= 1.0:
-                pass
-            else:
-                raise RelevanceScoreOutOfRangeException(relevance_score=self.relevance_score)
+            self._validate_relevance_score()
+
+    def _validate_relevance_score(self):
+        if 0.0 <= self.relevance_score <= 1.0:
+            pass
+        else:
+            raise RelevanceScoreOutOfRangeException(relevance_score=self.relevance_score)
 
     def dict(self) -> Dict[str, Any]:
         aps: Dict[str, Any] = {
@@ -54,3 +64,52 @@ def dict(self) -> Dict[str, Any]:
             payload.update(self.custom)
 
         return payload
+
+
+@dataclass(frozen=True)
+class LiveActivityPayload(Payload):
+    timestamp: datetime = field(default_factory=datetime.now)
+    event: LiveActivityEvent = field(default=None)
+    content_state: Dict[str, Any] = field(default_factory=dict)
+    stale_date: Optional[datetime] = field(default=None)
+    dismissal_date: Optional[datetime] = field(default=None)
+    attributes_type: Optional[str] = field(default=None)
+    attributes: Optional[Dict[str, Any]] = field(default=None)
+
+    def __post_init__(self):
+        if self.event is None:
+            raise ValueError("event must be specified.")
+        elif self.event == LiveActivityEvent.start:
+            if self.attributes_type is None or self.attributes is None:
+                raise ValueError(
+                    """attributes_type and attributes must be specified when event is start.
+                    Please see documentation: https://developer.apple.com/documentation/usernotifications/setting_up_a_remote_notification_server/generating_a_remote_notification"""  # noqa: E501
+                )
+            try:
+                _ = json.dumps(self.attributes)
+            except json.decoder.JSONDecodeError:
+                raise LiveActivityAttributesIsNotJSONSerializable()
+
+        try:
+            _ = json.dumps(self.content_state)
+        except json.decoder.JSONDecodeError:
+            raise LiveActivityContentStateIsNotJSONSerializable()
+        super().__post_init__()
+
+    def _validate_relevance_score(self):
+        # You can set any Double value; for example, 25, 50, 75, or 100.
+        pass
+
+    def dict(self) -> Dict[str, Any]:
+        payload = super().dict()
+        additional: Dict[str, Optional[Any]] = {
+            "timestamp": int(self.timestamp.timestamp()),
+            "event": self.event.value,
+            "content-state": self.content_state,
+            "state-date": int(self.stale_date.timestamp()) if self.stale_date else None,
+            "dismissal-date": int(self.dismissal_date.timestamp()) if self.dismissal_date else None,
+        }
+        additional = {k: v for k, v in additional.items() if v is not None}
+        payload["aps"].update(additional)
+
+        return payload

tests/clients/client/test_handle_error.py

@@ -164,7 +164,7 @@ def test_handle_error(reason, expect):
     client = __Client()
     exception = client._handle_error(error_json={"reason": reason})
 
-    assert type(exception) == expect
+    assert isinstance(exception, expect)
 
 
 def test_attributed_error():