Coverage for slidge/util/types.py: 97%
268 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-29 05:05 +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
146 date: datetime | None = None
147 sha256: str | None = None
148 """
149 Base64-encoded sha256 hash of the file. NB: not the hexdigest.
150 """
152 def __post_init__(self) -> None:
153 if all(
154 x is None
155 for x in (self.path, self.stream, self.data, self.url, self.aio_stream)
156 ):
157 raise TypeError("There is not data in this attachment", self)
159 if isinstance(self.path, str):
160 self.path = Path(self.path)
162 if self.is_sticker:
163 if self.disposition == "attachment":
164 warnings.warn(
165 "Sticker declared as 'attachment' disposition, changing it to 'inline'"
166 )
167 self.disposition = "inline"
169 def format_for_user(self) -> str:
170 if self.name:
171 name = self.name
172 elif self.path:
173 name = self.path.name # type:ignore[union-attr]
174 elif self.url:
175 name = self.url
176 else:
177 name = ""
179 if self.caption:
180 name = f"{name}: {self.caption}" if name else self.caption
182 return name
184 def __str__(self) -> str:
185 attrs = ", ".join(
186 f"{f.name}={getattr(self, f.name)!r}"
187 for f in fields(self)
188 if getattr(self, f.name) is not None and f.name != "data"
189 )
190 if self.data is not None:
191 data_str = f"data=<{len(self.data)} bytes>"
192 to_join = (attrs, data_str) if attrs else (data_str,)
193 attrs = ", ".join(to_join)
194 return f"Attachment({attrs})"
197class MucType(IntEnum):
198 """
199 The type of group, private, public, anonymous or not.
200 """
202 GROUP = 0
203 """
204 A private group, members-only and non-anonymous, eg a family group.
205 """
206 CHANNEL = 1
207 """
208 A public group, aka an anonymous channel.
209 """
210 CHANNEL_NON_ANONYMOUS = 2
211 """
212 A public group where participants' legacy IDs are visible to everybody.
213 """
216PseudoPresenceShow = PresenceShows | Literal[""]
219MessageOrPresenceTypeVar = TypeVar("MessageOrPresenceTypeVar", bound=Message | Presence)
222class LinkPreview(NamedTuple):
223 """
224 Embedded metadata from :xep:`0511`.
226 See <https://ogp.me/>_.
227 """
229 about: str
230 """
231 URL of the link.
232 """
233 title: str | None
234 """
235 Title of the linked page.
236 """
237 description: str | None
238 """
239 A description of the page.
240 """
241 url: str | None
242 """
243 The canonical URL of the link.
244 """
245 image: str | Path | bytes | None
246 """
247 An image representing the link. If it is a string, it should represent a URL to an image.
248 """
249 type: str | None
250 """
251 Type of the link destination.
252 """
253 site_name: str | None
254 """
255 Name of the web site.
256 """
258 @property
259 def is_empty(self) -> bool:
260 return not any(x for x in self)
263class Mention[LegacyParticipantType: AnyParticipant](NamedTuple):
264 participant: LegacyParticipantType
265 start: int
266 end: int
269class Hat(NamedTuple):
270 uri: str
271 title: str
272 hue: float | None = None
275class UserPreferences(TypedDict):
276 sync_avatar: bool
277 sync_presence: bool
280class MamMetadata(NamedTuple):
281 id: str
282 sent_on: datetime
285class HoleBound(NamedTuple):
286 id: str
287 timestamp: datetime
290class CachedPresence(NamedTuple):
291 last_seen: datetime | None = None
292 ptype: PresenceTypes | None = None
293 pstatus: str | None = None
294 pshow: PresenceShows | None = None
297class Avatar(NamedTuple):
298 path: Path | None = None
299 unique_id: str | None = None
300 url: str | None = None
301 data: bytes | None = None
304class AvatarMetadata(NamedTuple):
305 id: str
306 type: str
307 bytes: int
308 url: str
309 height: int
310 width: int
313class SpaceMetadata(NamedTuple):
314 creator_legacy_id: str | Unset | None = _UNSET
315 name: str | Unset | None = _UNSET
316 description: str | Unset | None = _UNSET
317 member_count: int | Unset | None = _UNSET
318 owner_legacy_ids: Iterable[str] | Unset = _UNSET
319 avatar: Avatar | Unset | None = _UNSET
320 banner: Avatar | Unset | None = _UNSET
323@dataclass
324class Reply:
325 """
326 Represents a message referenced (replied to) via :xep:`0461`
327 """
329 msg_id: str
330 """
331 The ID of the message being replied to.
332 """
333 fallback: str | None
334 """
335 A fallback (:xep:`0428`) text for clients not supporting :xep:`0461`,
336 usually consisting of the text of the referenced message.
337 NB: some XMPP clients might not fill this, so it can be empty.
338 """
339 to: Literal["self", "contact"] | LegacyParticipant[Any] = "contact"
340 """
341 Author of the referenced message.
342 """
344 @cached_property
345 def fallback_no_quote_mark(self) -> str | None:
346 """
347 Return multi-line text without leading quote marks (i.e. the ">" character).
348 """
349 if not self.fallback:
350 return None
351 return re.sub(_STRIP_QUOTE_RE, "", self.fallback).strip()
354_STRIP_QUOTE_RE = re.compile(r"^>\s*", flags=re.MULTILINE)
357class _ReplyProtocol(Protocol):
358 msg_id: str
359 fallback: str | None
360 to: Literal["self", "contact"] | LegacyParticipant[Any]
363@runtime_checkable
364class MUCReplyProtocol(_ReplyProtocol, Protocol):
365 to: LegacyParticipant[Any]
368@runtime_checkable
369class ContactReplyProtocol(_ReplyProtocol, Protocol):
370 to: Literal["self", "contact"]
373@dataclass
374class XMPPAttachment:
375 url: str
376 http: aiohttp.ClientSession | None = None
377 is_sticker: bool = False
378 cid: str | None = None
379 content_type: str | None = None
381 @contextlib.asynccontextmanager
382 async def get(self) -> AsyncIterator[aiohttp.ClientResponse]:
383 http_context = (
384 aiohttp.ClientSession()
385 if self.http is None
386 else contextlib.nullcontext(self.http)
387 )
388 async with http_context as http, http.get(self.url) as response:
389 yield response
392@dataclass
393class XMPPMessage[LegacyParticipantType: AnyParticipant]:
394 body: str | None = None
395 """
396 Text content of the message. Can be empty if there are attachments.
397 """
398 attachments: tuple[XMPPAttachment, ...] = ()
399 """
400 Attachments to this message (:xep:`0066`, :xep:`0385` or :xep:`0447`).
401 Is never empty if there is no body.
402 """
403 reply: Reply | None = None
404 """
405 A reference to message being replied-to (:xep:`0461`), if applicable.
406 """
407 link_previews: tuple[LinkPreview, ...] = ()
408 """
409 A list of link metadata (:xep:`0511`), attached to this message.
410 """
411 replace: str | None = None
412 """
413 The message this message is a correction for (:xep:`0308`).
414 """
415 thread: str | None = None
416 """
417 The thread this message is part of.
418 """
419 mentions: tuple[Mention[LegacyParticipantType], ...] = ()
420 """
421 A list of mentions parsed in the body of this message (by searching for
422 nicknames of MUC participants).
423 Is always empty for 1:1 messages.
424 """
427class XMPPMessageProtocol[LegacyParticipantType: AnyParticipant](Protocol):
428 body: str | None
429 attachments: tuple[XMPPAttachment, ...]
430 reply: _ReplyProtocol | None
431 link_previews: tuple[LinkPreview, ...]
432 replace: str | None
433 thread: str | None
434 mentions: tuple[Mention[LegacyParticipantType], ...]
437@runtime_checkable
438class ContactMessageProtocol(XMPPMessageProtocol[Any], Protocol):
439 reply: ContactReplyProtocol | None
440 mentions: tuple[()]
443@runtime_checkable
444class MUCMessageProtocol[LegacyParticipantType: AnyParticipant](
445 XMPPMessageProtocol[LegacyParticipantType], Protocol
446):
447 reply: MUCReplyProtocol | None
448 mentions: tuple[Mention[LegacyParticipantType], ...]
451class AttachmentMessageProtocol(XMPPMessageProtocol[LegacyParticipantType]):
452 attachments: tuple[XMPPAttachment, *tuple[XMPPAttachment, ...]]
455class XMPPTextMessageProtocol(XMPPMessageProtocol[LegacyParticipantType]):
456 body: str
457 attachments: tuple[()]
460class ContactAttachmentMessage(
461 ContactMessageProtocol, AttachmentMessageProtocol[LegacyParticipantType]
462):
463 pass
466class MUCAttachmentMessage[LegacyParticipantType: AnyParticipant](
467 MUCMessageProtocol[LegacyParticipantType],
468 AttachmentMessageProtocol[LegacyParticipantType],
469):
470 pass
473class ContactTextMessage(
474 ContactMessageProtocol, XMPPTextMessageProtocol[LegacyParticipantType]
475):
476 pass
479class MUCTextMessage[LegacyParticipantType: AnyParticipant](
480 MUCMessageProtocol[LegacyParticipantType],
481 XMPPTextMessageProtocol[LegacyParticipantType],
482):
483 pass
486MUCMessage = (
487 MUCAttachmentMessage[LegacyParticipantType] | MUCTextMessage[LegacyParticipantType]
488)
489ContactMessage = ContactAttachmentMessage[Any] | ContactTextMessage[Any]
492def has_attachments[LegacyParticipantType: AnyParticipant](
493 message: XMPPMessageProtocol[LegacyParticipantType],
494) -> TypeIs[AttachmentMessageProtocol[LegacyParticipantType]]:
495 return bool(message.attachments)
498@dataclass
499class Sticker:
500 path: Path
501 content_type: str | None
502 hashes: dict[str, str]
503 fallback: str | None = None
504 reply: Reply | None = None
505 thread: str | None = None
508class StickerProtocol(Protocol):
509 path: Path
510 content_type: str | None
511 hashes: dict[str, str]
512 fallback: str | None
513 reply: _ReplyProtocol | None
514 thread: str | None
517@runtime_checkable
518class ContactSticker(StickerProtocol, Protocol):
519 reply: ContactReplyProtocol | None
522@runtime_checkable
523class MUCSticker(StickerProtocol, Protocol):
524 reply: MUCReplyProtocol | None
527RegistrationValidationCoroutine = Callable[
528 [JID, "JSONSerializable"], Awaitable["JSONSerializable | None"]
529]