Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0465.py @ 4270:0d7bb4df2343

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Reformatted code base using black.
author Goffi <goffi@goffi.org>
date Wed, 19 Jun 2024 18:44:57 +0200
parents 4b842c1fb686
children
comparison
equal deleted inserted replaced
4269:64a85ce8be70 4270:0d7bb4df2343
118 subscriber_elt["jid"] = subscriber.full() 118 subscriber_elt["jid"] = subscriber.full()
119 return subscriber_elt 119 return subscriber_elt
120 120
121 @utils.ensure_deferred 121 @utils.ensure_deferred
122 async def _subscriptions( 122 async def _subscriptions(
123 self, 123 self, service="", nodeIdentifier="", profile_key=C.PROF_KEY_NONE
124 service="",
125 nodeIdentifier="",
126 profile_key=C.PROF_KEY_NONE
127 ) -> str: 124 ) -> str:
128 client = self.host.get_client(profile_key) 125 client = self.host.get_client(profile_key)
129 service = None if not service else jid.JID(service) 126 service = None if not service else jid.JID(service)
130 subs = await self.subscriptions(client, service, nodeIdentifier or None) 127 subs = await self.subscriptions(client, service, nodeIdentifier or None)
131 return data_format.serialise(subs) 128 return data_format.serialise(subs)
132 129
133 async def subscriptions( 130 async def subscriptions(
134 self, 131 self,
135 client: SatXMPPEntity, 132 client: SatXMPPEntity,
136 service: Optional[jid.JID] = None, 133 service: Optional[jid.JID] = None,
137 node: Optional[str] = None 134 node: Optional[str] = None,
138 ) -> List[Dict[str, Union[str, bool]]]: 135 ) -> List[Dict[str, Union[str, bool]]]:
139 """Retrieve public subscriptions from a service 136 """Retrieve public subscriptions from a service
140 137
141 @param service(jid.JID): PubSub service 138 @param service(jid.JID): PubSub service
142 @param nodeIdentifier(unicode, None): node to filter 139 @param nodeIdentifier(unicode, None): node to filter
168 "node": subscription_elt["node"], 165 "node": subscription_elt["node"],
169 "subscriber": service.full(), 166 "subscriber": service.full(),
170 "state": subscription_elt.getAttribute("subscription", "subscribed"), 167 "state": subscription_elt.getAttribute("subscription", "subscribed"),
171 } 168 }
172 except KeyError: 169 except KeyError:
173 log.warning( 170 log.warning(f"invalid <subscription> element: {subscription_elt.toXml()}")
174 f"invalid <subscription> element: {subscription_elt.toXml()}"
175 )
176 continue 171 continue
177 if node is not None and sub_dict["node"] != node: 172 if node is not None and sub_dict["node"] != node:
178 # if not is specified, we filter out any other node 173 # if not is specified, we filter out any other node
179 # FIXME: should node filtering be done by server? 174 # FIXME: should node filtering be done by server?
180 continue 175 continue
181 ret.append(sub_dict) 176 ret.append(sub_dict)
182 return ret 177 return ret
183 178
184 @utils.ensure_deferred 179 @utils.ensure_deferred
185 async def _get_public_node_subscriptions( 180 async def _get_public_node_subscriptions(
186 self, 181 self, service: str, node: str, profile_key: str
187 service: str,
188 node: str,
189 profile_key: str
190 ) -> Dict[str, str]: 182 ) -> Dict[str, str]:
191 client = self.host.get_client(profile_key) 183 client = self.host.get_client(profile_key)
192 subs = await self.get_public_node_subscriptions( 184 subs = await self.get_public_node_subscriptions(
193 client, jid.JID(service) if service else None, node 185 client, jid.JID(service) if service else None, node
194 ) 186 )
197 def get_public_subscribers_node(self, node: str) -> str: 189 def get_public_subscribers_node(self, node: str) -> str:
198 """Return prefixed node to retrieve public subscribers""" 190 """Return prefixed node to retrieve public subscribers"""
199 return f"{NS_PPS_SUBSCRIBERS}/{node}" 191 return f"{NS_PPS_SUBSCRIBERS}/{node}"
200 192
201 async def get_public_node_subscriptions( 193 async def get_public_node_subscriptions(
202 self, 194 self, client: SatXMPPEntity, service: Optional[jid.JID], nodeIdentifier: str
203 client: SatXMPPEntity,
204 service: Optional[jid.JID],
205 nodeIdentifier: str
206 ) -> Dict[jid.JID, str]: 195 ) -> Dict[jid.JID, str]:
207 """Retrieve public subscriptions to a node 196 """Retrieve public subscriptions to a node
208 197
209 @param nodeIdentifier(unicode): node to get subscriptions from 198 @param nodeIdentifier(unicode): node to get subscriptions from
210 """ 199 """
235 continue 224 continue
236 225
237 try: 226 try:
238 ret[jid.JID(subscriber_elt["jid"])] = "subscribed" 227 ret[jid.JID(subscriber_elt["jid"])] = "subscribed"
239 except (KeyError, RuntimeError): 228 except (KeyError, RuntimeError):
240 log.warning( 229 log.warning(f"invalid <subscriber> element: {subscriber_elt.toXml()}")
241 f"invalid <subscriber> element: {subscriber_elt.toXml()}"
242 )
243 continue 230 continue
244 return ret 231 return ret
245 232
246 def set_public_opt(self, options: Optional[dict] = None) -> dict: 233 def set_public_opt(self, options: Optional[dict] = None) -> dict:
247 """Set option to make a subscription public 234 """Set option to make a subscription public
251 238
252 @return: the options dict 239 @return: the options dict
253 """ 240 """
254 if options is None: 241 if options is None:
255 options = {} 242 options = {}
256 options[f'{{{NS_PPS}}}public'] = True 243 options[f"{{{NS_PPS}}}public"] = True
257 return options 244 return options
258 245
259 246
260 @implementer(iwokkel.IDisco) 247 @implementer(iwokkel.IDisco)
261 class XEP_0465_Handler(XMPPHandler): 248 class XEP_0465_Handler(XMPPHandler):