Coverage for src/lexigram/notification/backends/push/apns.py: 27%
83 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 07:17 +0800
1"""APNs (Apple Push Notification service) push notification backend."""
3from __future__ import annotations
5import asyncio
6import time
7from typing import TYPE_CHECKING
8import uuid
10from lexigram.contracts.core import HealthCheckResult, HealthStatus
11from lexigram.contracts.mailer import MessageDeliveryReceipt
12from lexigram.contracts.notification.types import PushMessage
13from lexigram.logging import get_logger
14from lexigram.notification.constants import (
15 APNS_BASE_URL,
16 APNS_SANDBOX_URL,
17 DEFAULT_APNS_TIMEOUT,
18)
19from lexigram.notification.exceptions import APNsNotificationError
20from lexigram.result import Err, Ok, Result
22if TYPE_CHECKING:
23 from lexigram.contracts.notification.errors import NotificationError
25logger = get_logger(__name__)
28class APNsPush:
29 """Apple Push Notification service backend.
31 Implements :class:`~lexigram.contracts.notification.protocols.PushChannelProtocol`.
32 Uses the APNs HTTP/2 REST API with JWT authentication.
33 Requires the ``httpx[http2]`` and ``PyJWT`` optional extras.
35 Args:
36 team_id: Apple Developer Team ID (10-character string).
37 key_id: APNs Auth Key ID (10-character string).
38 apns_auth_key: ECDSA private key — either a PEM string starting with
39 ``-----BEGIN PRIVATE KEY-----`` or a filesystem path to a ``.p8``
40 key file downloaded from the Apple Developer portal.
41 bundle_id: App bundle identifier (e.g. ``com.example.MyApp``).
42 Used as the ``apns-topic`` header value.
43 sandbox: Send via the APNs sandbox endpoint. Defaults to ``False``
44 (production).
45 timeout: HTTP request timeout in seconds. Defaults to 30.
46 """
48 def __init__(
49 self,
50 team_id: str,
51 key_id: str,
52 apns_auth_key: str,
53 bundle_id: str,
54 sandbox: bool = False,
55 timeout: int = DEFAULT_APNS_TIMEOUT,
56 ) -> None:
57 self._team_id = team_id
58 self._key_id = key_id
59 self._apns_auth_key = apns_auth_key
60 self._bundle_id = bundle_id
61 self._sandbox = sandbox
62 self._timeout = timeout
64 # ------------------------------------------------------------------
65 # Internal helpers
66 # ------------------------------------------------------------------
68 def _load_private_key(self) -> str:
69 """Return the raw PEM key string, loading from disk if necessary."""
70 stripped = self._apns_auth_key.strip()
71 if stripped.startswith("-----BEGIN"):
72 return stripped
73 # Treat as a filesystem path to a .p8 key file.
74 with open(stripped) as fh: # noqa: PTH123
75 return fh.read()
77 def _make_jwt(self) -> str:
78 """Generate a signed ES256 JWT token for APNs bearer authentication."""
79 import jwt
81 key_data = self._load_private_key()
82 issued_at = int(time.time())
83 payload = {"iss": self._team_id, "iat": issued_at}
84 # PyJWT merges extra_headers into the JWT header block.
85 return jwt.encode(
86 payload,
87 key_data,
88 algorithm="ES256",
89 headers={"kid": self._key_id},
90 )
92 async def _send_to_token(
93 self,
94 device_token: str,
95 message: PushMessage,
96 ) -> Result[MessageDeliveryReceipt, NotificationError]:
97 """Deliver one push notification to a single APNs device token."""
98 import httpx
100 base_url = APNS_SANDBOX_URL if self._sandbox else APNS_BASE_URL
101 url = f"{base_url}/3/device/{device_token}"
102 jwt_token = self._make_jwt()
104 headers = {
105 "authorization": f"bearer {jwt_token}",
106 "apns-topic": self._bundle_id,
107 "apns-push-type": "alert",
108 }
110 # Build the APS dictionary
111 aps: dict[str, object] = {
112 "alert": {
113 "title": message.title,
114 "body": message.body,
115 }
116 }
117 if message.badge is not None:
118 aps["badge"] = message.badge
119 if message.sound:
120 aps["sound"] = message.sound
122 # Root payload: aps + optional custom data keys
123 payload: dict[str, object] = {"aps": aps}
124 if message.data:
125 payload.update(message.data)
127 try:
128 async with httpx.AsyncClient(
129 http2=True,
130 timeout=float(self._timeout),
131 ) as client:
132 resp = await client.post(url, json=payload, headers=headers)
134 if resp.status_code == 200:
135 apns_id = resp.headers.get("apns-id")
136 receipt = MessageDeliveryReceipt(
137 message_id=str(uuid.uuid4()),
138 backend="apns",
139 channel="push",
140 provider_reference=apns_id,
141 )
142 logger.info(
143 "apns_sent",
144 token_prefix=device_token[:8],
145 apns_id=apns_id,
146 bundle_id=self._bundle_id,
147 sandbox=self._sandbox,
148 )
149 return Ok(receipt)
151 # Non-200: APNs returns JSON with a ``reason`` field.
152 try:
153 body = resp.json()
154 reason: str = body.get("reason", f"HTTP {resp.status_code}")
155 except Exception: # noqa: BLE001 # JSON parse failures are non-fatal
156 reason = f"HTTP {resp.status_code}"
158 logger.warning(
159 "apns_send_failed",
160 token_prefix=device_token[:8],
161 status=resp.status_code,
162 reason=reason,
163 sandbox=self._sandbox,
164 )
165 return Err(APNsNotificationError(reason, apns_reason=reason))
167 except OSError as exc:
168 logger.warning(
169 "apns_network_error",
170 token_prefix=device_token[:8],
171 error=str(exc),
172 )
173 return Err(APNsNotificationError(f"Network error: {exc}"))
175 # ------------------------------------------------------------------
176 # Public API (PushChannelProtocol)
177 # ------------------------------------------------------------------
179 async def send(
180 self,
181 message: PushMessage,
182 ) -> Result[MessageDeliveryReceipt, NotificationError]:
183 """Send a push notification via the APNs HTTP/2 API.
185 APNs accepts one device token per HTTP request. This method targets
186 ``message.to[0]``. Use :meth:`send_batch` to fan-out a single message
187 payload to multiple tokens concurrently.
189 Args:
190 message: The push notification to deliver. ``message.to`` must
191 contain at least one device token.
193 Returns:
194 ``Ok(MessageDeliveryReceipt)`` on HTTP 200.
195 ``Err(APNsNotificationError)`` for any APNs or network failure.
197 Raises:
198 ValueError: If ``message.to`` is empty.
199 """
200 if not message.to:
201 raise ValueError("APNsPush.send() requires at least one device token")
202 return await self._send_to_token(message.to[0], message)
204 async def send_batch(
205 self,
206 messages: list[PushMessage],
207 ) -> list[Result[MessageDeliveryReceipt, NotificationError]]:
208 """Send multiple push notifications concurrently.
210 Each :class:`~lexigram.contracts.notification.types.PushMessage` in
211 *messages* is dispatched to the first token in its ``to`` list.
213 Args:
214 messages: Notifications to deliver.
216 Returns:
217 ``Result`` list, one entry per message, order preserved.
218 """
219 tasks = [self.send(msg) for msg in messages]
220 return await asyncio.gather(*tasks)
222 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
223 """Verify APNs endpoint reachability via a HEAD request.
225 Args:
226 timeout: Maximum seconds to wait for a response.
228 Returns:
229 :class:`~lexigram.contracts.core.HealthCheckResult`.
230 """
231 import httpx
233 base_url = APNS_SANDBOX_URL if self._sandbox else APNS_BASE_URL
234 try:
235 async with httpx.AsyncClient(http2=True, timeout=timeout) as client:
236 resp = await client.head(base_url)
237 status = (
238 HealthStatus.HEALTHY
239 if resp.status_code < 500
240 else HealthStatus.UNHEALTHY
241 )
242 return HealthCheckResult(
243 component="apns",
244 status=status,
245 details={
246 "http_status": resp.status_code,
247 "sandbox": self._sandbox,
248 },
249 )
250 except OSError as exc:
251 return HealthCheckResult(
252 component="apns",
253 status=HealthStatus.UNHEALTHY,
254 details={"error": str(exc), "sandbox": self._sandbox},
255 )
258__all__ = ["APNsPush"]