1"""Internal utilities.
2
3:copyright: (c) 2024 Tanner Corcoran
4:license: Apache 2.0, see LICENSE for more details.
5
6"""
7
8import collections
9import copy
10import dataclasses
11import queue
12import sys
13import urllib.parse
14from collections.abc import Iterable, Mapping
15from types import MappingProxyType
16from typing import (
17 Annotated,
18 Any,
19 Callable,
20 TypeVar,
21 Union,
22 get_args,
23 get_origin,
24)
25
26if sys.version_info >= (3, 11): # pragma: no cover
27 from typing import Self, dataclass_transform
28else: # pragma: no cover
29 from typing_extensions import Self, dataclass_transform
30if sys.version_info >= (3, 10): # pragma: no cover
31 from typing import TypeAlias
32else: # pragma: no cover
33 from typing_extensions import TypeAlias
34
35from .__version__ import * # noqa: F401,F403
36
37_T = TypeVar("_T")
38StrTuple: TypeAlias = Union[tuple[str], tuple[str, ...]]
39
40
41_SECURE_URL_SCHEMES: set[str] = {
42 "https",
43 "wss",
44 "sftp",
45 "aaas",
46 "msrps",
47 "sips",
48}
49
50
51if sys.version_info >= (3, 10): # pragma: no cover
52
53 def _unwrap_static(wrapper: Callable[[Any], Any]) -> Callable[[Any], Any]:
54 """Get the callable from the given wrapper."""
55 return wrapper
56
57else: # pragma: no cover
58
59 def _unwrap_static(wrapper: Callable[[Any], Any]) -> Callable[[Any], Any]:
60 """Get the callable from the given wrapper."""
61 return (
62 wrapper.__func__ if isinstance(wrapper, staticmethod) else wrapper
63 )
64
65
[docs]
66@dataclasses.dataclass(eq=False, frozen=True)
67class URL:
68 """Internal URL handling for ntfy API endpoints.
69
70 :param scheme: The URL scheme (e.g., ``scheme`` in
71 ``scheme://netloc/path;parameters?query#fragment``).
72 :type scheme: str
73 :param netloc: The URL netloc (e.g., ``netloc`` in
74 ``scheme://netloc/path;parameters?query#fragment``).
75 :type netloc: str
76 :param path: The URL path (e.g., ``path`` in
77 ``scheme://netloc/path;parameters?query#fragment``).
78 :type path: str
79 :param params: The URL params (e.g., ``params`` in
80 ``scheme://netloc/path;parameters?query#fragment``).
81 :type params: str
82 :param query: The URL query (e.g., ``query`` in
83 ``scheme://netloc/path;parameters?query#fragment``).
84 :type query: str
85 :param fragment: The URL fragment (e.g., ``fragment`` in
86 ``scheme://netloc/path;parameters?query#fragment``).
87 :type fragment: str
88
89 """
90
91 scheme: str
92 """See the :paramref:`~URL.scheme` parameter."""
93
94 netloc: str
95 """See the :paramref:`~URL.netloc` parameter."""
96
97 path: str
98 """See the :paramref:`~URL.path` parameter."""
99
100 params: str
101 """See the :paramref:`~URL.params` parameter."""
102
103 query: str
104 """See the :paramref:`~URL.query` parameter."""
105
106 fragment: str
107 """See the :paramref:`~URL.fragment` parameter."""
108
[docs]
109 @classmethod
110 def parse(cls, url: str) -> Self:
111 """Parse a URL-like string into a new :class:`URL` instance.
112
113 :param url: The URL-like string to parse.
114 :type url: str
115
116 :return: A new :class:`URL` instance.
117 :rtype: URL
118
119 """
120 s, n, p, r, q, f = urllib.parse.urlparse(url)
121 return cls(s, n, p.rstrip("/"), r, q, f)
122
123 def _unparse(self, path: str, scheme: str) -> str:
124 """Internal method to reconstruct URL with a given path and
125 scheme.
126
127 :param path: The path to use in place of :attr:`.path`.
128 :type path: str
129 :param scheme: The path to use in place of :attr:`.scheme`.
130 :type scheme: str
131
132 :return: The reconstructed URL.
133 :rtype: str
134
135 """
136 return urllib.parse.urlunparse((
137 scheme,
138 self.netloc,
139 path,
140 self.params,
141 self.query,
142 self.fragment,
143 ))
144
[docs]
145 def unparse(
146 self,
147 endpoint: Union[str, Iterable[str], None] = None,
148 scheme: Union[tuple[str, str], None] = None,
149 ) -> str:
150 """Reconstruct the full URL string.
151
152 :param endpoint: An endpoint to be appended to the path before
153 parsing.
154 :type endpoint: str | typing.Iterable[str] | None,
155 optional
156 :param scheme: A scheme two-tuple (insecure, secure) to be used
157 instead of the existing scheme. Which version is used
158 (insecure vs secure) will be decided based on the current
159 scheme's security status.
160 :type scheme: tuple[str, str] | None, optional
161
162 :return: The constructed URL.
163 :rtype: str
164
165 """
166 if endpoint:
167 if isinstance(endpoint, str):
168 endpoint = (endpoint,)
169 e = "/".join(x.lstrip("/") for x in endpoint)
170 _path = f"{self.path}/{e}"
171 else:
172 _path = self.path
173 _scheme = (
174 scheme[self.scheme in _SECURE_URL_SCHEMES]
175 if scheme
176 else self.scheme
177 )
178
179 return self._unparse(_path, _scheme)
180
181
[docs]
182class ClearableQueue(queue.Queue[_T]):
183 """A :class:`queue.Queue` subclass that adds a single
184 :meth:`.clear` method, which clears all remaining items in the
185 queue.
186
187 """
188
189 queue: collections.deque[_T]
190
[docs]
191 def clear(self) -> None:
192 """Clear all remaining items in the queue."""
193 with self.mutex:
194 unfinished = self.unfinished_tasks - len(self.queue)
195
196 # finish unfinished tasks
197 if unfinished <= 0:
198 # would likely only happen if queue has been manually
199 # tampered with
200 if unfinished < 0:
201 raise ValueError("task_done() called too many times")
202
203 # notify threads waiting on the all_tasks_done threading
204 # condition
205 self.all_tasks_done.notify_all()
206
207 # updated unfinished tasks and clear queue
208 self.unfinished_tasks = unfinished
209 self.queue.clear()
210
211 # notify threads waiting on the not_full threading condition
212 self.not_full.notify_all()
213
214
[docs]
215@dataclass_transform()
216class WrappingDataclass:
217 """A special dataclass type that allows for its attributes to be
218 annotated with wrapper types.
219
220 """
221
222 _context: MappingProxyType[str, Callable[[Any], Any]]
223
224 def __init_subclass__(cls) -> None:
225 """Build context and initialize dataclass."""
226 cls._context = MappingProxyType({
227 k: c
228 for k, c in (
229 (k, cls._get_context(v))
230 for k, v in cls.__annotations__.items()
231 )
232 if c is not None
233 })
234 dataclasses.dataclass(cls)
235
236 @classmethod
237 def _get_context(
238 cls, annotation: type
239 ) -> Union[Callable[[Any], Any], None]:
240 """Get the context information from the given type annotation.
241
242 :param annotation: The annotation to get the context from.
243 :type annotation: type
244
245 :return: The context, if any.
246 :rtype: typing.Callable[[typing.Any], typing.Any] | None
247
248 """
249 origin = get_origin(annotation)
250
251 if origin is not Annotated:
252 return None
253
254 # typing.Annotated must have at least two arguments, so we know
255 # that index 1 will not be out of range
256 return get_args(annotation)[1]
257
[docs]
258 @classmethod
259 def from_json(cls, data: Mapping[str, Any]) -> Self:
260 """Parse a new :class:`.WrappingDataclass` instance from the
261 given data.
262
263 .. note:: :paramref:`.data` is not modified when wrappers (if
264 any) are applied. Instead, a shallow copy of the mapping is
265 created and used. Keep in mind that, because it is a shallow
266 copy, the wrappers may still modify the mapping values
267 in-place.
268
269 :param data: The JSON-like data.
270 :type data: typing.Mapping[str, typing.Any]
271
272 :return: The parsed :class:`.WrappingDataclass` instance.
273 :rtype: WrappingDataclass
274
275 """
276 _data = dict(copy.copy(data))
277 for k, v in data.items():
278 wrapper = cls._context.get(k)
279 if not wrapper:
280 continue
281 _data[k] = _unwrap_static(wrapper)(v)
282
283 return cls(**_data)