1"""The :class:`Filter` 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 Generator, Iterable
11from types import MappingProxyType
12from typing import Annotated, Any, Union, get_args
13
14# not 3.11 because we need frozen_default
15if sys.version_info >= (3, 12): # pragma: no cover
16 from typing import dataclass_transform
17else: # pragma: no cover
18 from typing_extensions import dataclass_transform
19
20from .__version__ import * # noqa: F401,F403
21
22__all__ = ("Filter",)
23
24
[docs]
25@dataclass_transform(frozen_default=True)
26class _Filter:
27 """A slightly modified version of :class:`._Message`
28 that is used to handle formatting.
29
30 """
31
32 _context: MappingProxyType[str, str]
33
34 def __init_subclass__(cls) -> None:
35 """Handle dataclass initialization and build the serialization
36 context.
37
38 """
39 cls._context = MappingProxyType(
40 {k: cls._get_context(a) for k, a in cls.__annotations__.items()}
41 )
42 dataclasses.dataclass(frozen=True)(cls)
43
44 @classmethod
45 def _get_context(cls, annotation: type) -> str:
46 """Get the context information from the given type annotation."""
47 return get_args(annotation)[1]
48
49 @classmethod
50 def _serializer(cls, value: Any) -> str:
51 """Serializer that does the following:
52
53 - Return the value if it is a string
54
55 - Stringify integers
56
57 - Format booleans as strings ("0" or "1")
58
59 - Iterables into a comma-separated list of its items (also
60 serialized)
61
62 """
63 if isinstance(value, str):
64 for o, n in (("\n", "n"), ("\r", "r"), ("\f", "f")):
65 value = value.replace(o, f"\\{n}")
66 return value
67
68 if isinstance(value, bool):
69 return ("0", "1")[value]
70
71 if isinstance(value, int):
72 return str(value)
73
74 try:
75 return ",".join(cls._serializer(e) for e in value)
76 except Exception:
77 pass
78
79 raise TypeError(f"Unknown type: {value.__class__.__name__!r}")
80
81 def _serialize(self) -> Generator[tuple[str, Any], None, None]:
82 """Generate segments that will later be turned into a dictionary
83 in :meth:`.serialize`.
84
85 """
86 for k, v in self.__dict__.items():
87 if v is None:
88 continue
89
90 yield (self._context[k], self._serializer(v))
91
[docs]
92 def serialize(self) -> dict[str, Any]:
93 """Serialize this filter into a header dictionary."""
94 return dict(self._serialize())
95
96
[docs]
97class Filter(_Filter):
98 """A filter dataclass that stores filtering preferences used when
99 polling and subscribing.
100
101 :param since: Return cached messages since timestamp, duration or
102 message ID.
103 :type since: str | int | None
104 :param scheduled: Include scheduled/delayed messages in message
105 response.
106 :type scheduled: bool | None
107 :param id: Only return messages that match this exact message ID.
108 :type id: str | None
109 :param message: Only return messages that match this exact message
110 string.
111 :type message: str | None
112 :param title: Only return messages that match this exact title
113 string.
114 :type title: str | None
115 :param priority: Only return messages that match **any** priority
116 given.
117 :type priority: int | typing.Iterable[int] | None
118 :param tags: Only return messages that match **all** listed tags.
119 :type tags: str | typing.Iterable[str] | None
120
121 """
122
123 since: Annotated[Union[str, int, None], "X-Since"] = None
124 """See the :paramref:`~Filter.since` parameter."""
125
126 scheduled: Annotated[Union[bool, None], "X-Scheduled"] = None
127 """See the :paramref:`~Filter.scheduled` parameter."""
128
129 id: Annotated[Union[str, None], "X-ID"] = None
130 """See the :paramref:`~Filter.id` parameter."""
131
132 message: Annotated[Union[str, None], "X-Message"] = None
133 """See the :paramref:`~Filter.message` parameter."""
134
135 title: Annotated[Union[str, None], "X-Title"] = None
136 """See the :paramref:`~Filter.title` parameter."""
137
138 priority: Annotated[Union[int, Iterable[int], None], "X-Priority"] = None
139 """See the :paramref:`~Filter.priority` parameter."""
140
141 tags: Annotated[Union[str, Iterable[str], None], "X-Tags"] = None
142 """See the :paramref:`~Filter.tags` parameter."""