1"""The `Message` class and its dependencies.
2
3:copyright: (c) 2024 Tanner Corcoran
4:license: Apache 2.0, see LICENSE for more details.
5
6"""
7
8import dataclasses
9import sys
10from collections.abc import Callable, Generator, Iterable
11from typing import Annotated, Any, BinaryIO, Protocol, Union, get_args
12
13# not 3.11 because we need frozen_default
14if sys.version_info >= (3, 12): # pragma: no cover
15 from typing import dataclass_transform
16else: # pragma: no cover
17 from typing_extensions import dataclass_transform
18
19from types import MappingProxyType
20
21from .__version__ import * # noqa: F401,F403
22from ._internals import WrappingDataclass, _unwrap_static
23from .actions import (
24 ReceivedAction,
25 ReceivedBroadcastAction,
26 ReceivedHTTPAction,
27 ReceivedViewAction,
28 _Action,
29)
30from .enums import Event, Priority, Tag
31
32__all__ = ("Message", "ReceivedAttachment", "ReceivedMessage")
33
34
[docs]
35@dataclass_transform(frozen_default=True)
36class _Message:
37 """Message base class that is used to handle formatting."""
38
39 _context: MappingProxyType[str, tuple[str, Callable[[Any], Any]]]
40
41 def __init_subclass__(cls) -> None:
42 """Handle dataclass initialization and build the serialization
43 context.
44
45 """
46 cls._context = MappingProxyType(
47 {k: cls._get_context(a) for k, a in cls.__annotations__.items()}
48 )
49 dataclasses.dataclass(frozen=True)(cls)
50
51 @classmethod
52 def _get_context(
53 cls, annotation: type
54 ) -> tuple[str, Callable[[Any], Any]]:
55 """Get the context information from the given type annotation."""
56 args = get_args(annotation)
57 serializer: Callable[[Any], Any] = cls._default_serializer
58
59 if len(args) == 3:
60 serializer = args[2]
61
62 return (args[1], serializer)
63
64 @staticmethod
65 def _default_serializer(value: Any) -> str:
66 """Default serializer that does the following:
67
68 - Return the value if it is a string
69
70 - Stringify integers
71
72 - Format booleans as strings
73
74 """
75 if isinstance(value, str):
76 for o, n in (("\n", "n"), ("\r", "r"), ("\f", "f")):
77 value = value.replace(o, f"\\{n}")
78 return value
79
80 if isinstance(value, bool):
81 return ("0", "1")[value]
82
83 if isinstance(value, int):
84 return str(value)
85
86 raise TypeError(f"Unknown type: {value.__class__.__name__!r}")
87
88 def _serialize(self) -> Generator[tuple[str, Any], None, None]:
89 """Generate segments that will later be turned into a dictionary
90 in :meth:`.serialize`.
91
92 """
93 for k, v in self.__dict__.items():
94 if v is None:
95 continue
96
97 key, serializer = self._context[k]
98
99 yield (key, _unwrap_static(serializer)(v))
100
[docs]
101 def serialize(self) -> dict[str, Any]:
102 """Serialize this message into a header dictionary."""
103 return dict(self._serialize())
104
105
[docs]
106class Message(_Message):
107 """Represents a message that will be published via
108 :meth:`.NtfyClient.publish`.
109
110 :param topic: The topic this message should be sent to.
111 :type topic: str
112 :param message: The message body.
113 :type message: str
114 :param title: The message title.
115 :type title: str
116 :param priority: The message priority.
117 :type priority: int
118 :param tags: A set of tags that will be displayed along with the
119 message. If the tag is a supported emoji short code (i.e.
120 those found in the :class:`.Tag` enum), they will be displayed
121 to the left of the title. Unrecognized tags will be displayed
122 below the message.
123 :type tags: str | typing.Iterable[str]
124 :param markdown: Whether or not to render the message with markdown.
125 :type markdown: bool
126 :param delay: When to schedule the message. This should be either a
127 unix timestamp or a natural-language date/time or offset
128 (see https://github.com/olebedev/when). Any value between 10
129 seconds and 3 days may be used.
130 :type delay: int | str
131 :param templating: If :py:obj:`True`, :paramref:`.data` will be
132 interpreted as JSON and used as the template context of
133 Go-styled templates in the :paramref:`.message` and
134 :paramref:`.title`. Mutually exclusive with
135 :paramref:`filename`.
136 :type templating: bool
137 :param actions: Actions attached to this message.
138 :type actions: typing.Iterable[_Action]
139 :param click: The redirect URL that will be activated when the
140 notification is clicked.
141 :type click: str
142 :param attachment: An externally-hosted attachment to be included in
143 the message.
144 :type attachment: str
145 :param filename: The filename of the included attachment. If not
146 provided, it will be derived from the URL resource
147 identifier. Mutually exclusive with :paramref:`templating`.
148 :type filename: str
149 :param icon: An externally-hosted icon that will be displayed next
150 to the notification.
151 :type icon: str
152 :param email: The email address to forward this message to.
153 :type email: str
154 :param call: The phone number to forward this message to.
155 :type call: str
156 :param cache: Whether or not to make use of the message cache.
157 :type cache: bool
158 :param firebase: Whether or not to enable Firebase.
159 :type firebase: bool
160 :param unified_push: Whether or not to enable UnifiedPush.
161 :type unified_push: bool
162 :param data: If defined, it should be one of the following:
163
164 * If sending a local file as an attachment, the raw
165 attachment file data. If doing so, it is recommended to
166 use the :paramref:`filename` option in addition to this.
167
168 * If using templating, the template data.
169
170 :type data: typing.BinaryIO | dict[str, typing.Any]
171
172 """
173
174 def __post_init__(self) -> None:
175 if self.templating is not None and self.filename is not None:
176 raise ValueError(
177 "The 'templating' and 'filename' options for 'Message' objects"
178 " are mutually exclusive"
179 )
180
181 @staticmethod
182 def _ignore(value: Any) -> Any:
183 return value
184
185 @staticmethod
186 def _tags_serializer(value: Union[str, Iterable[str]]) -> str:
187 if isinstance(value, str):
188 return value
189 return ",".join(value)
190
191 @staticmethod
192 def _actions_serializer(value: Iterable[_Action]) -> str:
193 return ";".join(v.serialize() for v in value)
194
195 topic: Annotated[Union[str, None], "__topic__"] = None
196 """See the :paramref:`~Message.topic` parameter."""
197
198 message: Annotated[Union[str, None], "X-Message"] = None
199 """See the :paramref:`~Message.message` parameter."""
200
201 title: Annotated[Union[str, None], "X-Title"] = None
202 """See the :paramref:`~Message.title` parameter."""
203
204 priority: Annotated[Union[int, None], "X-Priority"] = None
205 """See the :paramref:`~Message.priority` parameter."""
206
207 tags: Annotated[
208 Union[str, Iterable[str], None], "X-Tags", _tags_serializer
209 ] = None
210 """See the :paramref:`~Message.tags` parameter."""
211
212 markdown: Annotated[Union[bool, None], "X-Markdown"] = None
213 """See the :paramref:`~Message.markdown` parameter."""
214
215 delay: Annotated[Union[int, str, None], "X-Delay"] = None
216 """See the :paramref:`~Message.delay` parameter."""
217
218 templating: Annotated[Union[bool, None], "X-Template"] = None
219 """See the :paramref:`~Message.templating` parameter."""
220
221 actions: Annotated[
222 Union[Iterable[_Action], None], "X-Actions", _actions_serializer
223 ] = None
224 """See the :paramref:`~Message.actions` parameter."""
225
226 click: Annotated[Union[str, None], "X-Click"] = None
227 """See the :paramref:`~Message.click` parameter."""
228
229 attachment: Annotated[Union[str, None], "X-Attach"] = None
230 """See the :paramref:`~Message.attachment` parameter."""
231
232 filename: Annotated[Union[str, None], "X-Filename"] = None
233 """See the :paramref:`~Message.filename` parameter."""
234
235 icon: Annotated[Union[str, None], "X-Icon"] = None
236 """See the :paramref:`~Message.icon` parameter."""
237
238 email: Annotated[Union[str, None], "X-Email"] = None
239 """See the :paramref:`~Message.email` parameter."""
240
241 call: Annotated[Union[str, None], "X-Call"] = None
242 """See the :paramref:`~Message.call` parameter."""
243
244 cache: Annotated[Union[bool, None], "X-Cache"] = None
245 """See the :paramref:`~Message.cache` parameter."""
246
247 firebase: Annotated[Union[bool, None], "X-Firebase"] = None
248 """See the :paramref:`~Message.firebase` parameter."""
249
250 unified_push: Annotated[Union[bool, None], "X-UnifiedPush"] = None
251 """See the :paramref:`~Message.unified_push` parameter."""
252
253 data: Annotated[
254 Union[BinaryIO, dict[str, Any], None], "__data__", _ignore
255 ] = None
256 """See the :paramref:`~Message.data` parameter."""
257
[docs]
258 def get_args(
259 self,
260 ) -> tuple[Union[str, None], dict[str, str], dict[str, Any]]:
261 """Get the topic, headers, and POST kwargs."""
262 headers = self.serialize()
263 topic = headers.pop("__topic__", None)
264 data = headers.pop("__data__", None)
265
266 if data is None:
267 kwargs = {}
268 elif self.templating is True:
269 kwargs = {"json": data}
270 else:
271 kwargs = {"data": data}
272
273 return (topic, headers, kwargs)
274
275
[docs]
276class ReceivedAttachment(Protocol):
277 """Protocol defining the interface for an attachment in a received
278 message.
279
280 :param name: Name of the attachment.
281 :type name: str
282 :param url: URL of the attachment.
283 :type url: str
284 :param type: Mime type of the attachment (only if uploaded to ntfy
285 server).
286 :type type: str | None
287 :param size: Size in bytes (only if uploaded to ntfy server).
288 :type size: int | None
289 :param expires: Expiry date as Unix timestamp (only if uploaded to
290 ntfy server).
291 :type expires: int | None
292
293 """
294
295 name: str
296 """See the :paramref:`~ReceivedAttachment.name` parameter."""
297
298 url: str
299 """See the :paramref:`~ReceivedAttachment.url` parameter."""
300
301 type: Union[str, None]
302 """See the :paramref:`~ReceivedAttachment.type` parameter."""
303
304 size: Union[int, None]
305 """See the :paramref:`~ReceivedAttachment.size` parameter."""
306
307 expires: Union[int, None]
308 """See the :paramref:`~ReceivedAttachment.expires` parameter."""
309
310
[docs]
311class ReceivedMessage(Protocol):
312 """Protocol defining the interface for a message received from the
313 ntfy server.
314
315 :param id: Randomly chosen message identifier.
316 :type id: str
317 :param time: Message datetime as Unix timestamp.
318 :type time: int
319 :param event: Type of event.
320 :type event: Event
321 :param topic: Topics the message is associated with.
322 :type topic: str
323 :param message: Message body.
324 :type message: str | None
325 :param expires: When the message will be deleted.
326 :type expires: int | None
327 :param title: Message title.
328 :type title: str | None
329 :param tags: List of tags that may map to emojis.
330 :type tags: tuple[Tag, ...] | None
331 :param priority: Message priority.
332 :type priority: Priority | None
333 :param click: Website opened when notification is clicked.
334 :type click: str | None
335 :param actions: Action buttons that can be displayed.
336 :type actions: tuple[ReceivedAction, ...] | None
337 :param attachment: Details about an attachment if present.
338 :type attachment: ReceivedAttachment | None
339 :param icon: The icon URL sent with the message.
340 :type icon: str | None
341 :param content_type: The content type.
342 :type content_type: str | None
343
344 """
345
346 id: str
347 """See the :paramref:`~ReceivedMessage.id` parameter."""
348
349 time: int
350 """See the :paramref:`~ReceivedMessage.time` parameter."""
351
352 event: Event
353 """See the :paramref:`~ReceivedMessage.event` parameter."""
354
355 topic: str
356 """See the :paramref:`~ReceivedMessage.topic` parameter."""
357
358 message: Union[str, None]
359 """See the :paramref:`~ReceivedMessage.message` parameter."""
360
361 expires: Union[int, None]
362 """See the :paramref:`~ReceivedMessage.expires` parameter."""
363
364 title: Union[str, None]
365 """See the :paramref:`~ReceivedMessage.title` parameter."""
366
367 tags: Union[tuple[Tag, ...], None]
368 """See the :paramref:`~ReceivedMessage.tags` parameter."""
369
370 priority: Union[Priority, None]
371 """See the :paramref:`~ReceivedMessage.priority` parameter."""
372
373 click: Union[str, None]
374 """See the :paramref:`~ReceivedMessage.click` parameter."""
375
376 actions: Union[tuple[ReceivedAction, ...], None]
377 """See the :paramref:`~ReceivedMessage.actions` parameter."""
378
379 attachment: Union[ReceivedAttachment, None]
380 """See the :paramref:`~ReceivedMessage.attachment` parameter."""
381
382 icon: Union[str, None]
383 """See the :paramref:`~ReceivedMessage.icon` parameter."""
384
385 content_type: Union[str, None]
386 """See the :paramref:`~ReceivedMessage.content_type` parameter."""
387
388
[docs]
389class _ReceivedAttachment(ReceivedAttachment, WrappingDataclass):
390 """Private implementation of the :class:`ReceivedAttachment`
391 protocol.
392
393 :param name: Name of the attachment.
394 :type name: str
395 :param url: URL of the attachment.
396 :type url: str
397 :param type: Mime type of the attachment (only if uploaded to ntfy
398 server).
399 :type type: str | None
400 :param size: Size in bytes (only if uploaded to ntfy server).
401 :type size: int | None
402 :param expires: Expiry date as Unix timestamp (only if uploaded to
403 ntfy server).
404 :type expires: int | None
405
406 """
407
408 name: str
409 """See the :paramref:`~_ReceivedAttachment.name` parameter."""
410
411 url: str
412 """See the :paramref:`~_ReceivedAttachment.url` parameter."""
413
414 type: Union[str, None] = None
415 """See the :paramref:`~_ReceivedAttachment.type` parameter."""
416
417 size: Union[int, None] = None
418 """See the :paramref:`~_ReceivedAttachment.size` parameter."""
419
420 expires: Union[int, None] = None
421 """See the :paramref:`~_ReceivedAttachment.expires` parameter."""
422
423
[docs]
424class _ReceivedMessage(ReceivedMessage, WrappingDataclass):
425 """Private implementation of the :class:`ReceivedMessage` protocol.
426
427 :param id: Randomly chosen message identifier.
428 :type id: str
429 :param time: Message datetime as Unix timestamp.
430 :type time: int
431 :param event: Type of event.
432 :type event: Event
433 :param topic: Topics the message is associated with.
434 :type topic: str
435 :param message: Messag body.
436 :type message: str | None
437 :param expires: When the message will be deleted.
438 :type expires: int | None
439 :param title: Message title.
440 :type title: str | None
441 :param tags: List of tags that may map to emojis.
442 :type tags: tuple[Tag, ...] | None
443 :param priority: Message priority.
444 :type priority: Priority | None
445 :param click: Website opened when notification is clicked.
446 :type click: str | None
447 :param actions: Action buttons that can be displayed.
448 :type actions: tuple[ReceivedAction, ...] | None
449 :param attachment: Details about an attachment if present.
450 :type attachment: ReceivedAttachment | None
451 :param icon: The icon URL sent with the message.
452 :type icon: str | None
453 :param content_type: The content type.
454 :type content_type: str | None
455
456 """
457
458 @staticmethod
459 def _parse_actions(
460 actions: list[dict[str, Any]],
461 ) -> tuple[ReceivedAction, ...]:
462 return tuple(
463 dict[str, type[ReceivedAction]]({
464 "view": ReceivedViewAction,
465 "broadcast": ReceivedBroadcastAction,
466 "http": ReceivedHTTPAction,
467 })[a.pop("action")].from_json(a)
468 for a in actions
469 )
470
471 @staticmethod
472 def _parse_tags(tags: list[str]) -> tuple[Tag, ...]:
473 return tuple(map(Tag, tags))
474
475 id: str
476 """See the :paramref:`~_ReceivedMessage.id` parameter."""
477
478 time: int
479 """See the :paramref:`~_ReceivedMessage.time` parameter."""
480
481 event: Annotated[Event, Event]
482 """See the :paramref:`~_ReceivedMessage.event` parameter."""
483
484 topic: str
485 """See the :paramref:`~_ReceivedMessage.topic` parameter."""
486
487 message: Union[str, None] = None
488 """See the :paramref:`~_ReceivedMessage.message` parameter."""
489
490 expires: Union[int, None] = None
491 """See the :paramref:`~_ReceivedMessage.expires` parameter."""
492
493 title: Union[str, None] = None
494 """See the :paramref:`~_ReceivedMessage.title` parameter."""
495
496 tags: Annotated[Union[tuple[Tag, ...], None], _parse_tags] = None
497 """See the :paramref:`~_ReceivedMessage.tags` parameter."""
498
499 priority: Annotated[Union[Priority, None], Priority] = None
500 """See the :paramref:`~_ReceivedMessage.priority` parameter."""
501
502 click: Union[str, None] = None
503 """See the :paramref:`~_ReceivedMessage.click` parameter."""
504
505 actions: Annotated[
506 Union[tuple[ReceivedAction, ...], None], _parse_actions
507 ] = None
508 """See the :paramref:`~_ReceivedMessage.actions` parameter."""
509
510 attachment: Annotated[
511 Union[ReceivedAttachment, None], _ReceivedAttachment.from_json
512 ] = None
513 """See the :paramref:`~_ReceivedMessage.attachment` parameter."""
514
515 icon: Union[str, None] = None
516 """See the :paramref:`~_ReceivedMessage.icon` parameter."""
517
518 content_type: Union[str, None] = None
519 """See the :paramref:`~_ReceivedMessage.content_type` parameter."""