Merge pull request #183 from nnsnodnb/support-liveactivity

Support LiveActivity

README.md

@@ -53,13 +53,55 @@ asyncio.run(
 )
 ```
 
+### LiveActivity
+
+```python
+import asyncio
+from datetime import datetime
+
+from kalyke import LiveActivityClient, LiveActivityApnsConfig, LiveActivityEvent, LiveActivityPayload, PayloadAlert
+
+client = LiveActivityClient(
+    use_sandbox=True,
+    team_id="YOUR_TEAM_ID",
+    auth_key_id="AUTH_KEY_ID",
+    auth_key_filepath="/path/to/AuthKey_AUTH_KEY_ID.p8",
+)
+
+registration_id = "a8a799ba6c21e0795b07b577b562b8537418570c0fb8f7a64dca5a86a5a3b500"
+
+payload_alert = PayloadAlert(title="YOUR TITLE", body="YOUR BODY")
+payload = LiveActivityPayload(
+    alert=payload_alert,
+    badge=1,
+    sound="default",
+    timestamp=datetime.now(),
+    event=LiveActivityEvent.UPDATE,
+    content_state={
+        "currentHealthLevel": 100,
+        "eventDescription": "Adventure has begun!",
+    },
+)
+config = LiveActivityApnsConfig(
+    topic="com.example.App.push-type.liveactivity",
+)
+
+asyncio.run(
+    client.send_message(
+        device_token=registration_id,
+        payload=payload,
+        apns_config=config,
+    )
+)
+```
+
 ### VoIP
 
 ```python
 import asyncio
 from pathlib import Path
 
-from kalyke import ApnsConfig, ApnsPushType, VoIPClient
+from kalyke import VoIPApnsConfig, ApnsPushType, VoIPClient
 
 client = VoIPClient(
     use_sandbox=True,
@@ -69,9 +111,8 @@ client = VoIPClient(
 registration_id = "a8a799ba6c21e0795b07b577b562b8537418570c0fb8f7a64dca5a86a5a3b500"
 
 payload = {"key": "value"}
-config = ApnsConfig(
+config = VoIPApnsConfig(
     topic="com.example.App.voip",
-    push_type=ApnsPushType.VOIP,
 )
 
 asyncio.run(

kalyke/__init__.py

@@ -1,22 +1,34 @@
 from .clients.apns import ApnsClient
+from .clients.live_activity import LiveActivityClient
 from .clients.voip import VoIPClient
-from .models.apns_config import ApnsConfig
-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",
     "ApnsPriority",
     "ApnsPushType",
     "CriticalSound",
     "InterruptionLevel",
+    "LiveActivityApnsConfig",
+    "LiveActivityClient",
+    "LiveActivityEvent",
+    "LiveActivityPayload",
     "Payload",
     "PayloadAlert",
+    "VoIPClient",
+    "VoIPApnsConfig",
     "exceptions",
 ]

kalyke/clients/live_activity.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass
+from typing import Any, Dict, Union
+
+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: Union[LiveActivityPayload, Dict[str, Any]],
+        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,11 +1,11 @@
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Union
+from typing import Any, Dict, Union
 
 import httpx
 from httpx import AsyncClient
 
-from ..models import ApnsConfig
+from ..models import VoIPApnsConfig
 from . import __Client as BaseClient
 
 
@@ -21,7 +21,19 @@ def __post_init__(self):
         else:
             self._auth_key_filepath = Path(self.auth_key_filepath)
 
-    def _init_client(self, apns_config: ApnsConfig) -> AsyncClient:
+    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()
         context.load_cert_chain(self._auth_key_filepath)

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 "attributes is not JSON serializable."
+
+
+class LiveActivityContentStateIsNotJSONSerializable(Exception):
+    def __str__(self) -> str:
+        return "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/internal/status_code.py

@@ -1,7 +1,4 @@
 from enum import Enum
-from typing import TypeVar
-
-Self = TypeVar("Self", bound="StatusCode")
 
 
 class StatusCode(Enum):
@@ -19,39 +16,3 @@ class StatusCode(Enum):
     @property
     def is_success(self) -> bool:
         return self == self.SUCCESS
-
-    @property
-    def is_bad_request(self) -> bool:
-        return self == self.BAD_REQUEST
-
-    @property
-    def is_token_error(self) -> bool:
-        return self == self.TOKEN_ERROR
-
-    @property
-    def is_not_found(self) -> bool:
-        return self == self.NOT_FOUND
-
-    @property
-    def is_method_not_allowed(self) -> bool:
-        return self == self.METHOD_NOT_ALLOWED
-
-    @property
-    def is_token_inactive(self) -> bool:
-        return self == self.TOKEN_INACTIVE
-
-    @property
-    def is_payload_too_large(self) -> bool:
-        return self == self.PAYLOAD_TOO_LARGE
-
-    @property
-    def is_too_many_requests(self) -> bool:
-        return self == self.TOO_MARY_REQUESTS
-
-    @property
-    def is_internal_server_error(self) -> bool:
-        return self == self.INTERNAL_SERVER_ERROR
-
-    @property
-    def is_service_unavailable(self) -> bool:
-        return self == self.SERVER_UNAVAILABLE

kalyke/models/__init__.py

@@ -1,17 +1,22 @@
-from .apns_config import ApnsConfig
+from .apns_config import ApnsConfig, LiveActivityApnsConfig, VoIPApnsConfig
 from .apns_priority import ApnsPriority
 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__ = [
     "ApnsConfig",
+    "LiveActivityApnsConfig",
+    "VoIPApnsConfig",
     "ApnsPriority",
     "ApnsPushType",
     "CriticalSound",
     "InterruptionLevel",
+    "LiveActivityEvent",
     "Payload",
+    "LiveActivityPayload",
     "PayloadAlert",
 ]

kalyke/models/apns_config.py

@@ -37,3 +37,25 @@ def make_headers(self) -> Dict[str, str]:
         }
         attached_headers: Dict[str, str] = {k: v for k, v in headers.items() if v is not None}
         return attached_headers
+
+
+@dataclass(frozen=True)
+class LiveActivityApnsConfig(ApnsConfig):
+    push_type: ApnsPushType = field(default=ApnsPushType.LIVEACTIVITY, init=False)
+
+    def __post_init__(self):
+        if not self.topic.endswith(".push-type.liveactivity"):
+            raise ValueError(f"topic must end with .push-type.liveactivity, but {self.topic}.")
+        if self.priority == ApnsPriority.POWER_CONSIDERATIONS_OVER_ALL_OTHER_FACTORS:
+            raise ValueError("priority must be BASED_ON_POWER or IMMEDIATELY.")
+        super().__post_init__()
+
+
+@dataclass(frozen=True)
+class VoIPApnsConfig(ApnsConfig):
+    push_type: ApnsPushType = field(default=ApnsPushType.VOIP, init=False)
+
+    def __post_init__(self):
+        if not self.topic.endswith(".voip"):
+            raise ValueError(f"topic must end with .voip, but {self.topic}.")
+        super().__post_init__()

kalyke/models/apns_push_type.py

@@ -9,3 +9,4 @@ class ApnsPushType(Enum):
     COMPLICATION: str = "complication"
     FILE_PROVIDER: str = "fileprovider"
     MDM: str = "mdm"
+    LIVEACTIVITY: str = "liveactivity"

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,51 @@ 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.\nPlease 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 TypeError:
+                raise LiveActivityAttributesIsNotJSONSerializable()
+
+        try:
+            _ = json.dumps(self.content_state)
+        except TypeError:
+            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/apns/conftest.py

@@ -5,4 +5,4 @@
 
 @pytest.fixture()
 def auth_key_filepath() -> Path:
-    return Path(__file__).parent / "dummy.p8"
+    return Path(__file__).parent.parent / "dummy.p8"