Coverage for slidge / util / types.py: 95%

131 statements  

« prev     ^ index     » next       coverage.py v7.13.0, created at 2026-03-13 22:59 +0000

1""" 

2Typing stuff 

3""" 

4 

5from collections.abc import AsyncIterator, Hashable 

6from dataclasses import dataclass, fields 

7from datetime import datetime 

8from enum import IntEnum 

9from pathlib import Path 

10from typing import ( 

11 IO, 

12 TYPE_CHECKING, 

13 Any, 

14 Generic, 

15 Literal, 

16 NamedTuple, 

17 TypedDict, 

18 TypeVar, 

19 Union, 

20) 

21 

22from slixmpp import Message, Presence 

23from slixmpp.types import PresenceShows, PresenceTypes, ResourceDict # noqa: F401 

24 

25if TYPE_CHECKING: 

26 from ..contact import LegacyContact 

27 from ..core.session import BaseSession 

28 from ..group import LegacyMUC 

29 from ..group.participant import LegacyParticipant 

30 

31 AnyBaseSession = BaseSession[Any, Any] 

32else: 

33 # Hack to work around circular import. 

34 AnyBaseSession = int 

35 

36 

37LegacyGroupIdType = TypeVar("LegacyGroupIdType", bound=Hashable) 

38""" 

39Type of the unique identifier for groups, usually a str or an int, 

40but anything hashable should work. 

41""" 

42LegacyMessageType = TypeVar("LegacyMessageType", bound=Hashable) 

43LegacyThreadType = TypeVar("LegacyThreadType", bound=Hashable) 

44LegacyUserIdType = TypeVar("LegacyUserIdType", bound=Hashable) 

45 

46LegacyContactType = TypeVar("LegacyContactType", bound="LegacyContact[Any]") 

47LegacyMUCType = TypeVar("LegacyMUCType", bound="LegacyMUC[Any, Any, Any, Any]") 

48LegacyParticipantType = TypeVar("LegacyParticipantType", bound="LegacyParticipant") 

49 

50SessionType = TypeVar("SessionType", bound="BaseSession[Any, Any]") 

51Recipient = Union["LegacyMUC[Any, Any, Any, Any]", "LegacyContact[Any]"] 

52RecipientType = TypeVar("RecipientType", bound=Recipient) 

53Sender = Union["LegacyContact[Any]", "LegacyParticipant"] 

54LegacyFileIdType = int | str 

55 

56ChatState = Literal["active", "composing", "gone", "inactive", "paused"] 

57ProcessingHint = Literal["no-store", "markable", "store"] 

58Marker = Literal["acknowledged", "received", "displayed"] 

59FieldType = Literal[ 

60 "boolean", 

61 "fixed", 

62 "text-single", 

63 "text-multi", 

64 "jid-single", 

65 "jid-multi", 

66 "list-single", 

67 "list-multi", 

68 "text-private", 

69] 

70MucAffiliation = Literal["owner", "admin", "member", "outcast", "none"] 

71MucRole = Literal["visitor", "participant", "moderator", "none"] 

72# https://xmpp.org/registrar/disco-categories.html#client 

73ClientType = Literal[ 

74 "bot", "console", "game", "handheld", "pc", "phone", "sms", "tablet", "web" 

75] 

76AttachmentDisposition = Literal["attachment", "inline"] 

77 

78 

79@dataclass 

80class MessageReference(Generic[LegacyMessageType]): 

81 """ 

82 A "message reply", ie a "quoted message" (:xep:`0461`) 

83 

84 At the very minimum, the legacy message ID attribute must be set, but to 

85 ensure that the quote is displayed in all XMPP clients, the author must also 

86 be set (use the string "user" if the slidge user is the author of the referenced 

87 message). 

88 The body is used as a fallback for XMPP clients that do not support :xep:`0461` 

89 of that failed to find the referenced message. 

90 """ 

91 

92 legacy_id: LegacyMessageType 

93 author: Union[Literal["user"], "LegacyParticipant", "LegacyContact"] | None = None 

94 body: str | None = None 

95 

96 

97@dataclass 

98class LegacyAttachment: 

99 """ 

100 A file attachment to a message 

101 

102 At the minimum, one of the ``path``, ``steam``, ``data`` or ``url`` attribute 

103 has to be set 

104 

105 To be used with :meth:`.LegacyContact.send_files` or 

106 :meth:`.LegacyParticipant.send_files` 

107 """ 

108 

109 path: Path | str | None = None 

110 name: str | None = None 

111 stream: IO[bytes] | None = None 

112 aio_stream: AsyncIterator[bytes] | None = None 

113 data: bytes | None = None 

114 content_type: str | None = None 

115 legacy_file_id: str | int | None = None 

116 url: str | None = None 

117 caption: str | None = None 

118 disposition: AttachmentDisposition | None = None 

119 """ 

120 A caption for this specific image. For a global caption for a list of attachments, 

121 use the ``body`` parameter of :meth:`.AttachmentMixin.send_files` 

122 """ 

123 

124 def __post_init__(self) -> None: 

125 if all( 

126 x is None 

127 for x in (self.path, self.stream, self.data, self.url, self.aio_stream) 

128 ): 

129 raise TypeError("There is not data in this attachment", self) 

130 if isinstance(self.path, str): 

131 self.path = Path(self.path) 

132 

133 def format_for_user(self) -> str: 

134 if self.name: 

135 name = self.name 

136 elif self.path: 

137 name = self.path.name # type:ignore[union-attr] 

138 elif self.url: 

139 name = self.url 

140 else: 

141 name = "" 

142 

143 if self.caption: 

144 if name: 

145 name = f"{name}: {self.caption}" 

146 else: 

147 name = self.caption 

148 

149 return name 

150 

151 def __str__(self) -> str: 

152 attrs = ", ".join( 

153 f"{f.name}={getattr(self, f.name)!r}" 

154 for f in fields(self) 

155 if getattr(self, f.name) is not None and f.name != "data" 

156 ) 

157 if self.data is not None: 

158 data_str = f"data=<{len(self.data)} bytes>" 

159 to_join = (attrs, data_str) if attrs else (data_str,) 

160 attrs = ", ".join(to_join) 

161 return f"Attachment({attrs})" 

162 

163 

164class MucType(IntEnum): 

165 """ 

166 The type of group, private, public, anonymous or not. 

167 """ 

168 

169 GROUP = 0 

170 """ 

171 A private group, members-only and non-anonymous, eg a family group. 

172 """ 

173 CHANNEL = 1 

174 """ 

175 A public group, aka an anonymous channel. 

176 """ 

177 CHANNEL_NON_ANONYMOUS = 2 

178 """ 

179 A public group where participants' legacy IDs are visible to everybody. 

180 """ 

181 

182 

183PseudoPresenceShow = PresenceShows | Literal[""] 

184 

185 

186MessageOrPresenceTypeVar = TypeVar("MessageOrPresenceTypeVar", bound=Message | Presence) 

187 

188 

189class LinkPreview(NamedTuple): 

190 """ 

191 Embedded metadata from :xep:`0511`. 

192 

193 See <https://ogp.me/>_. 

194 """ 

195 

196 about: str 

197 """ 

198 URL of the link. 

199 """ 

200 title: str | None 

201 """ 

202 Title of the linked page. 

203 """ 

204 description: str | None 

205 """ 

206 A description of the page. 

207 """ 

208 url: str | None 

209 """ 

210 The canonical URL of the link. 

211 """ 

212 image: str | Path | bytes | None 

213 """ 

214 An image representing the link. If it is a string, it should represent a URL to an image. 

215 """ 

216 type: str | None 

217 """ 

218 Type of the link destination. 

219 """ 

220 site_name: str | None 

221 """ 

222 Name of the web site. 

223 """ 

224 

225 @property 

226 def is_empty(self) -> bool: 

227 return not any(x for x in self) 

228 

229 

230class Mention(NamedTuple): 

231 contact: "LegacyContact[Any]" 

232 start: int 

233 end: int 

234 

235 

236class Hat(NamedTuple): 

237 uri: str 

238 title: str 

239 hue: float | None = None 

240 

241 

242class UserPreferences(TypedDict): 

243 sync_avatar: bool 

244 sync_presence: bool 

245 

246 

247class MamMetadata(NamedTuple): 

248 id: str 

249 sent_on: datetime 

250 

251 

252class HoleBound(NamedTuple): 

253 id: int | str 

254 timestamp: datetime 

255 

256 

257class CachedPresence(NamedTuple): 

258 last_seen: datetime | None = None 

259 ptype: PresenceTypes | None = None 

260 pstatus: str | None = None 

261 pshow: PresenceShows | None = None 

262 

263 

264class Sticker(NamedTuple): 

265 path: Path 

266 content_type: str | None 

267 hashes: dict[str, str] 

268 

269 

270class Avatar(NamedTuple): 

271 path: Path | None = None 

272 unique_id: str | int | None = None 

273 url: str | None = None 

274 data: bytes | None = None