Coverage for slidge/group/bookmarks.py: 90%
101 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-04 08:17 +0000
1import abc
2import logging
3from typing import TYPE_CHECKING, Generic, Iterator, Optional, Type
5from slixmpp import JID
6from slixmpp.exceptions import XMPPError
8from ..db.models import Room
9from ..util import SubclassableOnce
10from ..util.jid_escaping import ESCAPE_TABLE, unescape_node
11from ..util.lock import NamedLockMixin
12from ..util.types import LegacyGroupIdType, LegacyMUCType
13from .room import LegacyMUC
15if TYPE_CHECKING:
16 from slidge.core.session import BaseSession
19class LegacyBookmarks(
20 Generic[LegacyGroupIdType, LegacyMUCType],
21 NamedLockMixin,
22 metaclass=SubclassableOnce,
23):
24 """
25 This is instantiated once per :class:`~slidge.BaseSession`
26 """
28 _muc_cls: Type[LegacyMUCType]
30 def __init__(self, session: "BaseSession") -> None:
31 self.session = session
32 self.xmpp = session.xmpp
33 self.user_jid = session.user_jid
35 self._user_nick: str = self.session.user_jid.node
37 super().__init__()
38 self.log = logging.getLogger(f"{self.user_jid.bare}:bookmarks")
39 self.ready = self.session.xmpp.loop.create_future()
40 if not self.xmpp.GROUPS:
41 self.ready.set_result(True)
43 @property
44 def user_nick(self):
45 return self._user_nick
47 @user_nick.setter
48 def user_nick(self, nick: str) -> None:
49 self._user_nick = nick
51 def from_store(self, stored: Room) -> LegacyMUCType:
52 return self._muc_cls(self.session, stored)
54 def __iter__(self) -> Iterator[LegacyMUC]:
55 with self.xmpp.store.session() as orm:
56 for stored in orm.query(Room).filter_by(user=self.session.user).all():
57 if stored.updated:
58 yield self.from_store(stored)
60 def __repr__(self) -> str:
61 return f"<Bookmarks of {self.user_jid}>"
63 async def legacy_id_to_jid_local_part(self, legacy_id: LegacyGroupIdType) -> str:
64 return await self.legacy_id_to_jid_username(legacy_id)
66 async def jid_local_part_to_legacy_id(self, local_part: str) -> LegacyGroupIdType:
67 return await self.jid_username_to_legacy_id(local_part)
69 async def legacy_id_to_jid_username(self, legacy_id: LegacyGroupIdType) -> str:
70 """
71 The default implementation calls ``str()`` on the legacy_id and
72 escape characters according to :xep:`0106`.
74 You can override this class and implement a more subtle logic to raise
75 an :class:`~slixmpp.exceptions.XMPPError` early
77 :param legacy_id:
78 :return:
79 """
80 return str(legacy_id).translate(ESCAPE_TABLE)
82 async def jid_username_to_legacy_id(self, username: str):
83 """
85 :param username:
86 :return:
87 """
88 return unescape_node(username)
90 async def by_jid(self, jid: JID) -> LegacyMUCType:
91 if jid.resource:
92 jid = JID(jid.bare)
93 async with self.lock(("bare", jid.bare)):
94 legacy_id = await self.jid_local_part_to_legacy_id(jid.node)
95 if self.get_lock(("legacy_id", legacy_id)):
96 self.session.log.debug("Already updating %s via by_legacy_id()", jid)
97 return await self.by_legacy_id(legacy_id)
99 with self.session.xmpp.store.session() as orm:
100 stored = (
101 orm.query(Room)
102 .filter_by(user_account_id=self.session.user_pk, jid=jid)
103 .one_or_none()
104 )
105 if stored is None:
106 stored = Room(
107 user_account_id=self.session.user_pk,
108 jid=jid,
109 legacy_id=legacy_id,
110 )
111 return await self.__update_if_needed(stored)
113 def by_jid_only_if_exists(self, jid: JID) -> Optional[LegacyMUCType]:
114 with self.xmpp.store.session() as orm:
115 stored = (
116 orm.query(Room).filter_by(user=self.session.user, jid=jid).one_or_none()
117 )
118 if stored is not None and stored.updated:
119 return self.from_store(stored)
120 return None
122 async def by_legacy_id(self, legacy_id: LegacyGroupIdType) -> LegacyMUCType:
123 async with self.lock(("legacy_id", legacy_id)):
124 local = await self.legacy_id_to_jid_local_part(legacy_id)
125 jid = JID(f"{local}@{self.xmpp.boundjid}")
126 if self.get_lock(("bare", jid.bare)):
127 self.session.log.debug("Already updating %s via by_jid()", jid)
128 return await self.by_jid(jid)
130 with self.xmpp.store.session() as orm:
131 stored = (
132 orm.query(Room)
133 .filter_by(
134 user_account_id=self.session.user_pk,
135 legacy_id=str(legacy_id),
136 )
137 .one_or_none()
138 )
139 if stored is None:
140 stored = Room(
141 user_account_id=self.session.user_pk,
142 jid=jid,
143 legacy_id=legacy_id,
144 )
145 return await self.__update_if_needed(stored)
147 async def __update_if_needed(self, stored: Room) -> LegacyMUCType:
148 muc = self.from_store(stored)
149 if muc.stored.updated:
150 return muc
152 with muc.updating_info():
153 try:
154 await muc.update_info()
155 except NotImplementedError:
156 pass
157 except XMPPError:
158 raise
159 except Exception as e:
160 raise XMPPError("internal-server-error", str(e))
162 return muc
164 @abc.abstractmethod
165 async def fill(self):
166 """
167 Establish a user's known groups.
169 This has to be overridden in plugins with group support and at the
170 minimum, this should ``await self.by_legacy_id(group_id)`` for all
171 the groups a user is part of.
173 Slidge internals will call this on successful :meth:`BaseSession.login`
175 """
176 if self.xmpp.GROUPS:
177 raise NotImplementedError(
178 "The plugin advertised support for groups but"
179 " LegacyBookmarks.fill() was not overridden."
180 )
182 async def remove(
183 self,
184 muc: LegacyMUC,
185 reason: str = "You left this group from the official client.",
186 kick: bool = True,
187 ) -> None:
188 """
189 Delete everything about a specific group.
191 This should be called when the user leaves the group from the official
192 app.
194 :param muc: The MUC to remove.
195 :param reason: Optionally, a reason why this group was removed.
196 :param kick: Whether the user should be kicked from this group. Set this
197 to False in case you do this somewhere else in your code, eg, on
198 receiving the confirmation that the group was deleted.
199 """
200 if kick:
201 user_participant = await muc.get_user_participant()
202 user_participant.kick(reason)
203 with self.xmpp.store.session() as orm:
204 orm.delete(muc.stored)
205 orm.commit()