Coverage for slidge/util/types.py: 97%
263 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-10 04:45 +0000
1"""
2Typing stuff
3"""
5from __future__ import annotations
7import contextlib
8import re
9import warnings
10from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
11from dataclasses import dataclass, fields
12from datetime import datetime
13from enum import IntEnum
14from functools import cached_property
15from pathlib import Path
16from typing import (
17 IO,
18 TYPE_CHECKING,
19 Any,
20 Literal,
21 NamedTuple,
22 Protocol,
23 TypedDict,
24 TypeIs,
25 TypeVar,
26 Union,
27 runtime_checkable,
28)
30import aiohttp
31from slixmpp import JID, Message, Presence
32from slixmpp.types import PresenceShows, PresenceTypes, ResourceDict # noqa: F401
34if TYPE_CHECKING:
35 from ..contact import LegacyContact, LegacyRoster
36 from ..core.gateway import BaseGateway
37 from ..core.session import BaseSession
38 from ..db.meta import JSONSerializable
39 from ..group import LegacyBookmarks, LegacyMUC
40 from ..group.participant import LegacyParticipant
42type AnySession = "BaseSession"
43type AnyGateway = "BaseGateway"
44type AnyMUC = "LegacyMUC[Any]"
45type AnyBookmarks = "LegacyBookmarks[Any]"
46type AnyRoster = "LegacyRoster[Any]"
47type AnyParticipant = "LegacyParticipant[Any]"
50class Unset:
51 def __bool__(self) -> Literal[False]:
52 return False
55_UNSET = Unset()
57LegacyContactType = TypeVar("LegacyContactType", bound="LegacyContact")
58LegacyMUCType = TypeVar("LegacyMUCType", bound=AnyMUC)
59LegacyParticipantType = TypeVar("LegacyParticipantType", bound=AnyParticipant)
60# Covariant with the bound as default, so that a bare `BaseSession` reference is typed
61# `BaseSession[LegacyRoster[Any], LegacyBookmarks[Any]]`` instead of `BaseSession[Any]`.
62LegacyRosterType_co = TypeVar(
63 "LegacyRosterType_co", bound=AnyRoster, default=AnyRoster, covariant=True
64)
65LegacyBookmarksType_co = TypeVar(
66 "LegacyBookmarksType_co", bound=AnyBookmarks, default=AnyBookmarks, covariant=True
67)
68SessionType_co = TypeVar(
69 "SessionType_co", bound=AnySession, default=AnySession, covariant=True
70)
72SessionType = TypeVar("SessionType", bound=AnySession)
73AnyRecipient = Union["LegacyContact", AnyMUC]
74RecipientType = TypeVar("RecipientType", bound=AnyRecipient)
75Sender = Union["LegacyContact", "AnyParticipant"]
77ChatState = Literal["active", "composing", "gone", "inactive", "paused"]
78ProcessingHint = Literal["no-store", "markable", "store"]
79Marker = Literal["acknowledged", "received", "displayed"]
80FieldType = Literal[
81 "boolean",
82 "fixed",
83 "text-single",
84 "text-multi",
85 "jid-single",
86 "jid-multi",
87 "list-single",
88 "list-multi",
89 "text-private",
90]
91MucAffiliation = Literal["owner", "admin", "member", "outcast", "none"]
92MucRole = Literal["visitor", "participant", "moderator", "none"]
93# https://xmpp.org/registrar/disco-categories.html#client
94ClientType = Literal[
95 "bot", "console", "game", "handheld", "pc", "phone", "sms", "tablet", "web"
96]
97AttachmentDisposition = Literal["attachment", "inline"]
100@dataclass
101class MessageReference:
102 """
103 A "message reply", ie a "quoted message" (:xep:`0461`)
105 At the very minimum, the legacy message ID attribute must be set, but to
106 ensure that the quote is displayed in all XMPP clients, the author must also
107 be set (use the string "user" if the slidge user is the author of the referenced
108 message).
109 The body is used as a fallback for XMPP clients that do not support :xep:`0461`
110 of that failed to find the referenced message.
111 """
113 legacy_id: str
114 author: Literal["user"] | AnyParticipant | LegacyContact | None = None
115 body: str | None = None
118@dataclass
119class LegacyAttachment:
120 """
121 A file attachment to a message
123 At the minimum, one of the ``path``, ``steam``, ``data`` or ``url`` attribute
124 has to be set
126 To be used with :meth:`.LegacyContact.send_files` or
127 :meth:`.LegacyParticipant.send_files`
128 """
130 path: Path | str | None = None
131 name: str | None = None
132 stream: IO[bytes] | None = None
133 aio_stream: AsyncIterator[bytes] | None = None
134 data: bytes | None = None
135 content_type: str | None = None
136 legacy_file_id: str | None = None
137 url: str | None = None
138 caption: str | None = None
139 """
140 A caption for this specific image. For a global caption for a list of attachments,
141 use the ``body`` parameter of :meth:`.AttachmentMixin.send_files`
142 """
143 disposition: AttachmentDisposition | None = None
144 is_sticker: bool = False
145 size: int | None = None
147 def __post_init__(self) -> None:
148 if all(
149 x is None
150 for x in (self.path, self.stream, self.data, self.url, self.aio_stream)
151 ):
152 raise TypeError("There is not data in this attachment", self)
154 if isinstance(self.path, str):
155 self.path = Path(self.path)
157 if self.is_sticker:
158 if self.disposition == "attachment":
159 warnings.warn(
160 "Sticker declared as 'attachment' disposition, changing it to 'inline'"
161 )
162 self.disposition = "inline"
164 def format_for_user(self) -> str:
165 if self.name:
166 name = self.name
167 elif self.path:
168 name = self.path.name # type:ignore[union-attr]
169 elif self.url:
170 name = self.url
171 else:
172 name = ""
174 if self.caption:
175 name = f"{name}: {self.caption}" if name else self.caption
177 return name
179 def __str__(self) -> str:
180 attrs = ", ".join(
181 f"{f.name}={getattr(self, f.name)!r}"
182 for f in fields(self)
183 if getattr(self, f.name) is not None and f.name != "data"
184 )
185 if self.data is not None:
186 data_str = f"data=<{len(self.data)} bytes>"
187 to_join = (attrs, data_str) if attrs else (data_str,)
188 attrs = ", ".join(to_join)
189 return f"Attachment({attrs})"
192class MucType(IntEnum):
193 """
194 The type of group, private, public, anonymous or not.
195 """
197 GROUP = 0
198 """
199 A private group, members-only and non-anonymous, eg a family group.
200 """
201 CHANNEL = 1
202 """
203 A public group, aka an anonymous channel.
204 """
205 CHANNEL_NON_ANONYMOUS = 2
206 """
207 A public group where participants' legacy IDs are visible to everybody.
208 """
211PseudoPresenceShow = PresenceShows | Literal[""]
214MessageOrPresenceTypeVar = TypeVar("MessageOrPresenceTypeVar", bound=Message | Presence)
217class LinkPreview(NamedTuple):
218 """
219 Embedded metadata from :xep:`0511`.
221 See <https://ogp.me/>_.
222 """
224 about: str
225 """
226 URL of the link.
227 """
228 title: str | None
229 """
230 Title of the linked page.
231 """
232 description: str | None
233 """
234 A description of the page.
235 """
236 url: str | None
237 """
238 The canonical URL of the link.
239 """
240 image: str | Path | bytes | None
241 """
242 An image representing the link. If it is a string, it should represent a URL to an image.
243 """
244 type: str | None
245 """
246 Type of the link destination.
247 """
248 site_name: str | None
249 """
250 Name of the web site.
251 """
253 @property
254 def is_empty(self) -> bool:
255 return not any(x for x in self)
258class Mention[LegacyParticipantType: AnyParticipant](NamedTuple):
259 participant: LegacyParticipantType
260 start: int
261 end: int
264class Hat(NamedTuple):
265 uri: str
266 title: str
267 hue: float | None = None
270class UserPreferences(TypedDict):
271 sync_avatar: bool
272 sync_presence: bool
275class MamMetadata(NamedTuple):
276 id: str
277 sent_on: datetime
280class HoleBound(NamedTuple):
281 id: str
282 timestamp: datetime
285class CachedPresence(NamedTuple):
286 last_seen: datetime | None = None
287 ptype: PresenceTypes | None = None
288 pstatus: str | None = None
289 pshow: PresenceShows | None = None
292class Avatar(NamedTuple):
293 path: Path | None = None
294 unique_id: str | None = None
295 url: str | None = None
296 data: bytes | None = None
299class AvatarMetadata(NamedTuple):
300 id: str
301 type: str
302 bytes: int
303 url: str
304 height: int
305 width: int
308class SpaceMetadata(NamedTuple):
309 creator_legacy_id: str | Unset | None = _UNSET
310 name: str | Unset | None = _UNSET
311 description: str | Unset | None = _UNSET
312 member_count: int | Unset | None = _UNSET
313 owner_legacy_ids: Iterable[str] | Unset = _UNSET
314 avatar: Avatar | Unset | None = _UNSET
315 banner: Avatar | Unset | None = _UNSET
318@dataclass
319class Reply:
320 """
321 Represents a message referenced (replied to) via :xep:`0461`
322 """
324 msg_id: str
325 """
326 The ID of the message being replied to.
327 """
328 fallback: str | None
329 """
330 A fallback (:xep:`0428`) text for clients not supporting :xep:`0461`,
331 usually consisting of the text of the referenced message.
332 NB: some XMPP clients might not fill this, so it can be empty.
333 """
334 to: Literal["self", "contact"] | LegacyParticipant[Any] = "contact"
335 """
336 Author of the referenced message.
337 """
339 @cached_property
340 def fallback_no_quote_mark(self) -> str | None:
341 """
342 Return multi-line text without leading quote marks (i.e. the ">" character).
343 """
344 if not self.fallback:
345 return None
346 return re.sub(_STRIP_QUOTE_RE, "", self.fallback).strip()
349_STRIP_QUOTE_RE = re.compile(r"^>\s*", flags=re.MULTILINE)
352class _ReplyProtocol(Protocol):
353 msg_id: str
354 fallback: str | None
355 to: Literal["self", "contact"] | LegacyParticipant[Any]
358@runtime_checkable
359class MUCReplyProtocol(_ReplyProtocol, Protocol):
360 to: LegacyParticipant[Any]
363@runtime_checkable
364class ContactReplyProtocol(_ReplyProtocol, Protocol):
365 to: Literal["self", "contact"]
368@dataclass
369class XMPPAttachment:
370 url: str
371 is_sticker: bool = False
372 cid: str | None = None
373 content_type: str | None = None
375 @contextlib.asynccontextmanager
376 async def get(self) -> AsyncIterator[aiohttp.ClientResponse]:
377 async with (
378 aiohttp.ClientSession() as session,
379 session.get(self.url) as response,
380 ):
381 yield response
384@dataclass
385class XMPPMessage[LegacyParticipantType: AnyParticipant]:
386 body: str | None = None
387 """
388 Text content of the message. Can be empty if there are attachments.
389 """
390 attachments: tuple[XMPPAttachment, ...] = ()
391 """
392 Attachments to this message (:xep:`0066`, :xep:`0385` or :xep:`0447`).
393 Is never empty if there is no body.
394 """
395 reply: Reply | None = None
396 """
397 A reference to message being replied-to (:xep:`0461`), if applicable.
398 """
399 link_previews: tuple[LinkPreview, ...] = ()
400 """
401 A list of link metadata (:xep:`0511`), attached to this message.
402 """
403 replace: str | None = None
404 """
405 The message this message is a correction for (:xep:`0308`).
406 """
407 thread: str | None = None
408 """
409 The thread this message is part of.
410 """
411 mentions: tuple[Mention[LegacyParticipantType], ...] = ()
412 """
413 A list of mentions parsed in the body of this message (by searching for
414 nicknames of MUC participants).
415 Is always empty for 1:1 messages.
416 """
419class XMPPMessageProtocol[LegacyParticipantType: AnyParticipant](Protocol):
420 body: str | None
421 attachments: tuple[XMPPAttachment, ...]
422 reply: _ReplyProtocol | None
423 link_previews: tuple[LinkPreview, ...]
424 replace: str | None
425 thread: str | None
426 mentions: tuple[Mention[LegacyParticipantType], ...]
429@runtime_checkable
430class ContactMessageProtocol(XMPPMessageProtocol[Any], Protocol):
431 reply: ContactReplyProtocol | None
432 mentions: tuple[()]
435@runtime_checkable
436class MUCMessageProtocol[LegacyParticipantType: AnyParticipant](
437 XMPPMessageProtocol[LegacyParticipantType], Protocol
438):
439 reply: MUCReplyProtocol | None
440 mentions: tuple[Mention[LegacyParticipantType], ...]
443class AttachmentMessageProtocol(XMPPMessageProtocol[LegacyParticipantType]):
444 attachments: tuple[XMPPAttachment, *tuple[XMPPAttachment, ...]]
447class XMPPTextMessageProtocol(XMPPMessageProtocol[LegacyParticipantType]):
448 body: str
449 attachments: tuple[()]
452class ContactAttachmentMessage(
453 ContactMessageProtocol, AttachmentMessageProtocol[LegacyParticipantType]
454):
455 pass
458class MUCAttachmentMessage[LegacyParticipantType: AnyParticipant](
459 MUCMessageProtocol[LegacyParticipantType],
460 AttachmentMessageProtocol[LegacyParticipantType],
461):
462 pass
465class ContactTextMessage(
466 ContactMessageProtocol, XMPPTextMessageProtocol[LegacyParticipantType]
467):
468 pass
471class MUCTextMessage[LegacyParticipantType: AnyParticipant](
472 MUCMessageProtocol[LegacyParticipantType],
473 XMPPTextMessageProtocol[LegacyParticipantType],
474):
475 pass
478MUCMessage = (
479 MUCAttachmentMessage[LegacyParticipantType] | MUCTextMessage[LegacyParticipantType]
480)
481ContactMessage = ContactAttachmentMessage[Any] | ContactTextMessage[Any]
484def has_attachments[LegacyParticipantType: AnyParticipant](
485 message: XMPPMessageProtocol[LegacyParticipantType],
486) -> TypeIs[AttachmentMessageProtocol[LegacyParticipantType]]:
487 return bool(message.attachments)
490@dataclass
491class Sticker:
492 path: Path
493 content_type: str | None
494 hashes: dict[str, str]
495 fallback: str | None = None
496 reply: Reply | None = None
497 thread: str | None = None
500class StickerProtocol(Protocol):
501 path: Path
502 content_type: str | None
503 hashes: dict[str, str]
504 fallback: str | None
505 reply: _ReplyProtocol | None
506 thread: str | None
509@runtime_checkable
510class ContactSticker(StickerProtocol, Protocol):
511 reply: ContactReplyProtocol | None
514@runtime_checkable
515class MUCSticker(StickerProtocol, Protocol):
516 reply: MUCReplyProtocol | None
519RegistrationValidationCoroutine = Callable[
520 [JID, "JSONSerializable"], Awaitable["JSONSerializable | None"]
521]