Coverage for slidge/db/models.py: 98%

1import warnings 

2from datetime import datetime 

3from enum import IntEnum 

4from typing import Optional 

5 

6import sqlalchemy as sa 

7from slixmpp import JID 

8from slixmpp.types import MucAffiliation, MucRole 

9from sqlalchemy import ForeignKey, Index, UniqueConstraint 

10from sqlalchemy.orm import Mapped, mapped_column, relationship 

11 

12from ..util.types import ClientType, MucType 

13from .meta import Base, JSONSerializable, JSONSerializableTypes 

14 

15 

16class XmppToLegacyEnum(IntEnum): 

17 """ 

18 XMPP-client generated IDs, used in the XmppToLegacyIds table to keep track 

19 of corresponding legacy IDs 

20 """ 

21 

22 DM = 1 

23 GROUP_CHAT = 2 

24 THREAD = 3 

25 

26 

27class ArchivedMessageSource(IntEnum): 

28 """ 

29 Whether an archived message comes from ``LegacyMUC.backfill()`` or was received 

30 as a "live" message. 

31 """ 

32 

33 LIVE = 1 

34 BACKFILL = 2 

35 

36 

37class GatewayUser(Base): 

38 """ 

39 A user, registered to the gateway component. 

40 """ 

41 

42 __tablename__ = "user_account" 

43 id: Mapped[int] = mapped_column(primary_key=True) 

44 jid: Mapped[JID] = mapped_column(unique=True) 

45 registration_date: Mapped[datetime] = mapped_column( 

46 sa.DateTime, server_default=sa.func.now() 

47 ) 

48 

49 legacy_module_data: Mapped[JSONSerializable] = mapped_column(default={}) 

50 """ 

51 Arbitrary non-relational data that legacy modules can use 

52 """ 

53 preferences: Mapped[JSONSerializable] = mapped_column(default={}) 

54 avatar_hash: Mapped[Optional[str]] = mapped_column(default=None) 

55 """ 

56 Hash of the user's avatar, to avoid re-publishing the same avatar on the 

57 legacy network 

58 """ 

59 

60 contacts: Mapped[list["Contact"]] = relationship( 

61 back_populates="user", cascade="all, delete-orphan" 

62 ) 

63 rooms: Mapped[list["Room"]] = relationship( 

64 back_populates="user", cascade="all, delete-orphan" 

65 ) 

66 xmpp_to_legacy: Mapped[list["XmppToLegacyIds"]] = relationship( 

67 cascade="all, delete-orphan" 

68 ) 

69 attachments: Mapped[list["Attachment"]] = relationship(cascade="all, delete-orphan") 

70 multi_legacy: Mapped[list["LegacyIdsMulti"]] = relationship( 

71 cascade="all, delete-orphan" 

72 ) 

73 multi_xmpp: Mapped[list["XmppIdsMulti"]] = relationship( 

74 cascade="all, delete-orphan" 

75 ) 

76 

77 def __repr__(self) -> str: 

78 return f"User(id={self.id!r}, jid={self.jid!r})" 

79 

80 def get(self, field: str, default: str = "") -> JSONSerializableTypes: 

81 # """ 

82 # Get fields from the registration form (required to comply with slixmpp backend protocol) 

83 # 

84 # :param field: Name of the field 

85 # :param default: Default value to return if the field is not present 

86 # 

87 # :return: Value of the field 

88 # """ 

89 return self.legacy_module_data.get(field, default) 

90 

91 @property 

92 def registration_form(self) -> dict: 

93 # Kept for retrocompat, should be 

94 # FIXME: delete me 

95 warnings.warn( 

96 "GatewayUser.registration_form is deprecated.", DeprecationWarning 

97 ) 

98 return self.legacy_module_data 

99 

100 

101class Avatar(Base): 

102 """ 

103 Avatars of contacts, rooms and participants. 

104 

105 To comply with XEPs, we convert them all to PNG before storing them. 

106 """ 

107 

108 __tablename__ = "avatar" 

109 

110 id: Mapped[int] = mapped_column(primary_key=True) 

111 

112 filename: Mapped[str] = mapped_column(unique=True) 

113 hash: Mapped[str] = mapped_column(unique=True) 

114 height: Mapped[int] = mapped_column() 

115 width: Mapped[int] = mapped_column() 

116 

117 # this is only used when avatars are available as HTTP URLs and do not 

118 # have a legacy_id 

119 url: Mapped[Optional[str]] = mapped_column(default=None) 

120 etag: Mapped[Optional[str]] = mapped_column(default=None) 

121 last_modified: Mapped[Optional[str]] = mapped_column(default=None) 

122 

123 contacts: Mapped[list["Contact"]] = relationship(back_populates="avatar") 

124 rooms: Mapped[list["Room"]] = relationship(back_populates="avatar") 

125 

126 

127class Contact(Base): 

128 """ 

129 Legacy contacts 

130 """ 

131 

132 __tablename__ = "contact" 

133 __table_args__ = ( 

134 UniqueConstraint("user_account_id", "legacy_id"), 

135 UniqueConstraint("user_account_id", "jid"), 

136 ) 

137 

138 id: Mapped[int] = mapped_column(primary_key=True) 

139 user_account_id: Mapped[int] = mapped_column(ForeignKey("user_account.id")) 

140 user: Mapped[GatewayUser] = relationship(back_populates="contacts") 

141 legacy_id: Mapped[str] = mapped_column(nullable=False) 

142 

143 jid: Mapped[JID] = mapped_column() 

144 

145 avatar_id: Mapped[int] = mapped_column(ForeignKey("avatar.id"), nullable=True) 

146 avatar: Mapped[Avatar] = relationship(back_populates="contacts") 

147 

148 nick: Mapped[Optional[str]] = mapped_column(nullable=True) 

149 

150 cached_presence: Mapped[bool] = mapped_column(default=False) 

151 last_seen: Mapped[Optional[datetime]] = mapped_column(nullable=True) 

152 ptype: Mapped[Optional[str]] = mapped_column(nullable=True) 

153 pstatus: Mapped[Optional[str]] = mapped_column(nullable=True) 

154 pshow: Mapped[Optional[str]] = mapped_column(nullable=True) 

155 caps_ver: Mapped[Optional[str]] = mapped_column(nullable=True) 

156 

157 is_friend: Mapped[bool] = mapped_column(default=False) 

158 added_to_roster: Mapped[bool] = mapped_column(default=False) 

159 sent_order: Mapped[list["ContactSent"]] = relationship( 

160 back_populates="contact", cascade="all, delete-orphan" 

161 ) 

162 

163 extra_attributes: Mapped[Optional[JSONSerializable]] = mapped_column( 

164 default=None, nullable=True 

165 ) 

166 updated: Mapped[bool] = mapped_column(default=False) 

167 

168 vcard: Mapped[Optional[str]] = mapped_column() 

169 vcard_fetched: Mapped[bool] = mapped_column(default=False) 

170 

171 participants: Mapped[list["Participant"]] = relationship(back_populates="contact") 

172 

173 avatar_legacy_id: Mapped[Optional[str]] = mapped_column(nullable=True) 

174 

175 client_type: Mapped[ClientType] = mapped_column(nullable=False, default="pc") 

176 

177 

178class ContactSent(Base): 

179 """ 

180 Keep track of XMPP msg ids sent by a specific contact for networks in which 

181 all messages need to be marked as read. 

182 

183 (XMPP displayed markers convey a "read up to here" semantic.) 

184 """ 

185 

186 __tablename__ = "contact_sent" 

187 __table_args__ = (UniqueConstraint("contact_id", "msg_id"),) 

188 

189 id: Mapped[int] = mapped_column(primary_key=True) 

190 contact_id: Mapped[int] = mapped_column(ForeignKey("contact.id")) 

191 contact: Mapped[Contact] = relationship(back_populates="sent_order") 

192 msg_id: Mapped[str] = mapped_column() 

193 

194 

195class Room(Base): 

196 """ 

197 Legacy room 

198 """ 

199 

200 __table_args__ = ( 

201 UniqueConstraint( 

202 "user_account_id", "legacy_id", name="uq_room_user_account_id_legacy_id" 

203 ), 

204 UniqueConstraint("user_account_id", "jid", name="uq_room_user_account_id_jid"), 

205 ) 

206 

207 __tablename__ = "room" 

208 id: Mapped[int] = mapped_column(primary_key=True) 

209 user_account_id: Mapped[int] = mapped_column(ForeignKey("user_account.id")) 

210 user: Mapped[GatewayUser] = relationship(back_populates="rooms") 

211 legacy_id: Mapped[str] = mapped_column(nullable=False) 

212 

213 jid: Mapped[JID] = mapped_column(nullable=False) 

214 

215 avatar_id: Mapped[int] = mapped_column(ForeignKey("avatar.id"), nullable=True) 

216 avatar: Mapped[Avatar] = relationship(back_populates="rooms") 

217 

218 name: Mapped[Optional[str]] = mapped_column(nullable=True) 

219 description: Mapped[Optional[str]] = mapped_column(nullable=True) 

220 subject: Mapped[Optional[str]] = mapped_column(nullable=True) 

221 subject_date: Mapped[Optional[datetime]] = mapped_column(nullable=True) 

222 subject_setter: Mapped[Optional[str]] = mapped_column(nullable=True) 

223 

224 n_participants: Mapped[Optional[int]] = mapped_column(default=None) 

225 

226 muc_type: Mapped[Optional[MucType]] = mapped_column(default=MucType.GROUP) 

227 

228 user_nick: Mapped[Optional[str]] = mapped_column() 

229 user_resources: Mapped[Optional[str]] = mapped_column(nullable=True) 

230 

231 participants_filled: Mapped[bool] = mapped_column(default=False) 

232 history_filled: Mapped[bool] = mapped_column(default=False) 

233 

234 extra_attributes: Mapped[Optional[JSONSerializable]] = mapped_column(default=None) 

235 updated: Mapped[bool] = mapped_column(default=False) 

236 

237 participants: Mapped[list["Participant"]] = relationship( 

238 back_populates="room", 

239 primaryjoin="Participant.room_id == Room.id", 

240 cascade="all, delete-orphan", 

241 ) 

242 

243 avatar_legacy_id: Mapped[Optional[str]] = mapped_column(nullable=True) 

244 

245 archive: Mapped[list["ArchivedMessage"]] = relationship( 

246 cascade="all, delete-orphan" 

247 ) 

248 

249 

250class ArchivedMessage(Base): 

251 """ 

252 Messages of rooms, that we store to act as a MAM server 

253 """ 

254 

255 __tablename__ = "mam" 

256 __table_args__ = (UniqueConstraint("room_id", "stanza_id"),) 

257 

258 id: Mapped[int] = mapped_column(primary_key=True) 

259 room_id: Mapped[int] = mapped_column(ForeignKey("room.id"), nullable=False) 

260 

261 stanza_id: Mapped[str] = mapped_column(nullable=False) 

262 timestamp: Mapped[datetime] = mapped_column(nullable=False) 

263 author_jid: Mapped[JID] = mapped_column(nullable=False) 

264 source: Mapped[ArchivedMessageSource] = mapped_column(nullable=False) 

265 legacy_id: Mapped[Optional[str]] = mapped_column(nullable=True) 

266 

267 stanza: Mapped[str] = mapped_column(nullable=False) 

268 

269 

270class XmppToLegacyIds(Base): 

271 """ 

272 XMPP-client generated IDs, and mapping to the corresponding legacy IDs 

273 """ 

274 

275 __tablename__ = "xmpp_to_legacy_ids" 

276 __table_args__ = ( 

277 Index("xmpp_legacy", "user_account_id", "xmpp_id", "legacy_id", unique=True), 

278 ) 

279 id: Mapped[int] = mapped_column(primary_key=True) 

280 user_account_id: Mapped[int] = mapped_column(ForeignKey("user_account.id")) 

281 user: Mapped[GatewayUser] = relationship(back_populates="xmpp_to_legacy") 

282 

283 xmpp_id: Mapped[str] = mapped_column(nullable=False) 

284 legacy_id: Mapped[str] = mapped_column(nullable=False) 

285 

286 type: Mapped[XmppToLegacyEnum] = mapped_column(nullable=False) 

287 

288 

289class Attachment(Base): 

290 """ 

291 Legacy attachments 

292 """ 

293 

294 __tablename__ = "attachment" 

295 

296 id: Mapped[int] = mapped_column(primary_key=True) 

297 user_account_id: Mapped[int] = mapped_column(ForeignKey("user_account.id")) 

298 user: Mapped[GatewayUser] = relationship(back_populates="attachments") 

299 

300 legacy_file_id: Mapped[Optional[str]] = mapped_column(index=True, nullable=True) 

301 url: Mapped[str] = mapped_column(index=True, nullable=False) 

302 sims: Mapped[Optional[str]] = mapped_column() 

303 sfs: Mapped[Optional[str]] = mapped_column() 

304 

305 

306class LegacyIdsMulti(Base): 

307 """ 

308 Legacy messages with multiple attachments are split as several XMPP messages, 

309 this table and the next maps a single legacy ID to multiple XMPP IDs. 

310 """ 

311 

312 __tablename__ = "legacy_ids_multi" 

313 __table_args__ = ( 

314 Index( 

315 "legacy_ids_multi_user_account_id_legacy_id", 

316 "user_account_id", 

317 "legacy_id", 

318 unique=True, 

319 ), 

320 ) 

321 id: Mapped[int] = mapped_column(primary_key=True) 

322 user_account_id: Mapped[int] = mapped_column(ForeignKey("user_account.id")) 

323 

324 legacy_id: Mapped[str] = mapped_column(nullable=False) 

325 xmpp_ids: Mapped[list["XmppIdsMulti"]] = relationship( 

326 back_populates="legacy_ids_multi", cascade="all, delete-orphan" 

327 ) 

328 

329 

330class XmppIdsMulti(Base): 

331 __tablename__ = "xmpp_ids_multi" 

332 __table_args__ = ( 

333 Index( 

334 "legacy_ids_multi_user_account_id_xmpp_id", 

335 "user_account_id", 

336 "xmpp_id", 

337 unique=True, 

338 ), 

339 ) 

340 id: Mapped[int] = mapped_column(primary_key=True) 

341 user_account_id: Mapped[int] = mapped_column(ForeignKey("user_account.id")) 

342 

343 xmpp_id: Mapped[str] = mapped_column(nullable=False) 

344 

345 legacy_ids_multi_id: Mapped[int] = mapped_column(ForeignKey("legacy_ids_multi.id")) 

346 legacy_ids_multi: Mapped[LegacyIdsMulti] = relationship(back_populates="xmpp_ids") 

347 

348 

349participant_hats = sa.Table( 

350 "participant_hats", 

351 Base.metadata, 

352 sa.Column("participant_id", ForeignKey("participant.id"), primary_key=True), 

353 sa.Column("hat_id", ForeignKey("hat.id"), primary_key=True), 

354) 

355 

356 

357class Hat(Base): 

358 __tablename__ = "hat" 

359 __table_args__ = (UniqueConstraint("title", "uri"),) 

360 

361 id: Mapped[int] = mapped_column(primary_key=True) 

362 title: Mapped[str] = mapped_column() 

363 uri: Mapped[str] = mapped_column() 

364 participants: Mapped[list["Participant"]] = relationship( 

365 secondary=participant_hats, back_populates="hats" 

366 ) 

367 

368 

369class Participant(Base): 

370 __tablename__ = "participant" 

371 

372 id: Mapped[int] = mapped_column(primary_key=True) 

373 

374 room_id: Mapped[int] = mapped_column(ForeignKey("room.id"), nullable=False) 

375 room: Mapped[Room] = relationship( 

376 back_populates="participants", primaryjoin=Room.id == room_id 

377 ) 

378 

379 contact_id: Mapped[int] = mapped_column(ForeignKey("contact.id"), nullable=True) 

380 contact: Mapped[Contact] = relationship(lazy=False, back_populates="participants") 

381 

382 is_user: Mapped[bool] = mapped_column(default=False) 

383 

384 affiliation: Mapped[MucAffiliation] = mapped_column(default="member") 

385 role: Mapped[MucRole] = mapped_column(default="participant") 

386 

387 presence_sent: Mapped[bool] = mapped_column(default=False) 

388 

389 resource: Mapped[Optional[str]] = mapped_column(default=None) 

390 nickname: Mapped[str] = mapped_column(nullable=True, default=None) 

391 

392 hats: Mapped[list["Hat"]] = relationship( 

393 secondary=participant_hats, back_populates="participants" 

394 ) 

395 

396 extra_attributes: Mapped[Optional[JSONSerializable]] = mapped_column(default=None) 

397 

398 

399class Bob(Base): 

400 __tablename__ = "bob" 

401 

402 id: Mapped[int] = mapped_column(primary_key=True) 

403 file_name: Mapped[str] = mapped_column(nullable=False) 

404 

405 sha_1: Mapped[str] = mapped_column(nullable=False, unique=True) 

406 sha_256: Mapped[str] = mapped_column(nullable=False, unique=True) 

407 sha_512: Mapped[str] = mapped_column(nullable=False, unique=True) 

408 

409 content_type: Mapped[Optional[str]] = mapped_column(nullable=False)