libervia-backend: libervia/backend/plugins/plugin_pubsub

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

Reformatted code base using black.

author	Goffi <goffi@goffi.org>
date	Wed, 19 Jun 2024 18:44:57 +0200
parents	ba46d6a0ff3a
children

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 CACHE_LIMIT = 5000
 # number of second before a progress caching is considered failed and tried again
 PROGRESS_DEADLINE = 60 * 60 * 6
 class PubsubCache:
 # TODO: there is currently no notification for (un)subscribe events with XEP-0060,
 #   but it would be necessary to have this data if some devices unsubscribe a cached
 #   node, as we can then get out of sync. A protoXEP could be proposed to fix this
 #   situation.
 raise ValueError('"name" is mandatory in analyser')
 if "type" not in analyser:
 raise ValueError('"type" is mandatory in analyser')
 type_test_keys = {"node", "namespace"}
 if not type_test_keys.intersection(analyser):
-raise ValueError(f'at least one of {type_test_keys} must be used')
+raise ValueError(f"at least one of {type_test_keys} must be used")
 if name in self.analysers:
 raise exceptions.Conflict(
 f"An analyser with the name {name!r} is already registered"
 )
 self.analysers[name] = analyser
 async def cache_items(
-self,
+self, client: SatXMPPEntity, pubsub_node: PubsubNode, items: List[domish.Element]
-client: SatXMPPEntity,
-pubsub_node: PubsubNode,
-items: List[domish.Element]
 ) -> None:
 try:
 parser = self.analysers[pubsub_node.analyser].get("parser")
 except KeyError:
 parser = None
 if parser is not None:
 parsed_items = [
 await utils.as_deferred(
-parser,
+parser, client, item, pubsub_node.service, pubsub_node.name
-client,
-item,
-pubsub_node.service,
-pubsub_node.name
 )
 for item in items
 ]
 else:
 parsed_items = None
 await self.host.memory.storage.cache_pubsub_items(
 client, pubsub_node, items, parsed_items
 )
-async def _cache_node(
+async def _cache_node(self, client: SatXMPPEntity, pubsub_node: PubsubNode) -> None:
-self,
-client: SatXMPPEntity,
-pubsub_node: PubsubNode
-) -> None:
 await self.host.memory.storage.update_pubsub_node_sync_state(
 pubsub_node, SyncState.IN_PROGRESS
 )
 service, node = pubsub_node.service, pubsub_node.name
 try:
-log.debug(
+log.debug(f"Caching node {node!r} at {service} for {client.profile}")
-f"Caching node {node!r} at {service} for {client.profile}"
-)
 if not pubsub_node.subscribed:
 try:
 sub = await self._p.subscribe(client, service, node)
 except Exception as e:
 log.warning(
 "latest 20 items"
 )
 items, __ = await client.pubsub_client.items(
 pubsub_node.service, pubsub_node.name, maxItems=20
 )
-await self.cache_items(
+await self.cache_items(client, pubsub_node, items)
-client, pubsub_node, items
-)
 else:
 raise e
 except exceptions.FeatureNotFound:
 log.warning(
 f"service {service} doesn't handle Result Set Management "
 "(XEP-0059), we'll only cache latest 20 items"
 )
 items, __ = await client.pubsub_client.items(
 pubsub_node.service, pubsub_node.name, maxItems=20
 )
-await self.cache_items(
+await self.cache_items(client, pubsub_node, items)
-client, pubsub_node, items
-)
 else:
 rsm_p = self.host.plugins["XEP-0059"]
 rsm_request = rsm.RSMRequest()
 cached_ids = set()
 while True:
 items, rsm_response = await client.pubsub_client.items(
 service, node, rsm_request=rsm_request
 )
-await self.cache_items(
+await self.cache_items(client, pubsub_node, items)
-client, pubsub_node, items
-)
 for item in items:
 item_id = item["id"]
 if item_id in cached_ids:
 log.warning(
 f"Pubsub node {node!r} at {service} is returning several "
 await self.host.memory.storage.update_pubsub_node_sync_state(
 pubsub_node, SyncState.COMPLETED
 )
 except Exception as e:
 import traceback
 tb = traceback.format_tb(e.__traceback__)
 log.error(
 f"Can't cache node {node!r} at {service} for {client.profile}: {e}\n{tb}"
 )
 await self.host.memory.storage.update_pubsub_node_sync_state(
 raise e
 def _cache_node_clean(self, __, pubsub_node):
 del self.in_progress[(pubsub_node.service, pubsub_node.name)]
-def cache_node(
+def cache_node(self, client: SatXMPPEntity, pubsub_node: PubsubNode) -> None:
-self,
-client: SatXMPPEntity,
-pubsub_node: PubsubNode
-) -> None:
 """Launch node caching as a background task"""
 d = defer.ensureDeferred(self._cache_node(client, pubsub_node))
 d.addBoth(self._cache_node_clean, pubsub_node=pubsub_node)
 self.in_progress[(pubsub_node.service, pubsub_node.name)] = d
 return d
 async def analyse_node(
 self,
 client: SatXMPPEntity,
 service: jid.JID,
 node: str,
-pubsub_node : PubsubNode = None,
+pubsub_node: PubsubNode = None,
 ) -> dict:
 """Use registered analysers on a node to determine what it is used for"""
 analyse = {"service": service, "node": node}
 if pubsub_node is None:
 try:
-first_item = (await client.pubsub_client.items(
+first_item = (await client.pubsub_client.items(service, node, 1))[0][0]
-service, node, 1
-))[0][0]
 except IndexError:
 pass
 except error.StanzaError as e:
 if e.condition == "item-not-found":
 pass
 found = True
 break
 else:
 found = False
-log.debug(
+log.debug(f"node {node!r} at service {service} doesn't match any known type")
-f"node {node!r} at service {service} doesn't match any known type"
-)
 if found:
 try:
 match_cb = analyser["match_cb"]
 except KeyError:
 pass
 else:
 await utils.as_deferred(match_cb, client, analyse)
 return analyse
 def _get_items_from_cache(
-self, service="", node="", max_items=10, item_ids=None, sub_id=None,
+self,
-extra="", profile_key=C.PROF_KEY_NONE
+service="",
+node="",
+max_items=10,
+item_ids=None,
+sub_id=None,
+extra="",
+profile_key=C.PROF_KEY_NONE,
 ):
-d = defer.ensureDeferred(self._a_get_items_from_cache(
+d = defer.ensureDeferred(
-service, node, max_items, item_ids, sub_id, extra, profile_key
+self._a_get_items_from_cache(
-))
+service, node, max_items, item_ids, sub_id, extra, profile_key
+)
+)
 d.addCallback(self._p.trans_items_data)
 d.addCallback(self._p.serialise_items)
 return d
 async def _a_get_items_from_cache(
 node: PubsubNode,
 max_items: Optional[int] = None,
 item_ids: Optional[List[str]] = None,
 sub_id: Optional[str] = None,
 rsm_request: Optional[rsm.RSMRequest] = None,
-extra: Optional[Dict[str, Any]] = None
+extra: Optional[Dict[str, Any]] = None,
 ) -> Tuple[List[PubsubItem], dict]:
 """Get items from cache, using same arguments as for external Pubsub request"""
 if extra is None:
 extra = {}
 if "mam" in extra:
 raise NotImplementedError("MAM queries are not supported yet")
 if max_items is None and rsm_request is None:
 max_items = 20
 pubsub_items, metadata = await self.host.memory.storage.get_items(
-node, max_items=max_items, item_ids=item_ids or None,
+node,
-order_by=extra.get(C.KEY_ORDER_BY)
+max_items=max_items,
+item_ids=item_ids or None,
+order_by=extra.get(C.KEY_ORDER_BY),
 )
 elif max_items is not None:
 if rsm_request is not None:
 raise exceptions.InternalError(
 "Pubsub max items and RSM must not be used at the same time"
 before = None
 desc = True
 else:
 before = rsm_request.before
 pubsub_items, metadata = await self.host.memory.storage.get_items(
-node, max_items=rsm_request.max, before=before, after=rsm_request.after,
+node,
-from_index=rsm_request.index, order_by=extra.get(C.KEY_ORDER_BY),
+max_items=rsm_request.max,
-desc=desc, force_rsm=True,
+before=before,
+after=rsm_request.after,
+from_index=rsm_request.index,
+order_by=extra.get(C.KEY_ORDER_BY),
+desc=desc,
+force_rsm=True,
 )
 return pubsub_items, metadata
 async def on_items_event(self, client, event):
 log.warning(
 f"Unexpected Pubsub event element: {xml_tools.p_fmt_elt(elt)}"
 )
 if items:
 log.debug(f"[{client.profile}] caching new items received from {node}")
-await self.cache_items(
+await self.cache_items(client, node, items)
-client, node, items
-)
 if retract_ids:
 log.debug(f"deleting retracted items from {node}")
 await self.host.memory.storage.delete_pubsub_items(
 node, items_names=retract_ids
 )
 node: str,
 max_items: Optional[int],
 item_ids: Optional[List[str]],
 sub_id: Optional[str],
 rsm_request: Optional[rsm.RSMRequest],
-extra: dict
+extra: dict,
 ) -> Tuple[bool, Optional[Tuple[List[dict], dict]]]:
 if not self.use_cache:
 log.debug("cache disabled in settings")
 return True, None
 if extra.get(C.KEY_USE_CACHE) == False:
 client: SatXMPPEntity,
 service: jid.JID,
 nodeIdentifier: str,
 sub_jid: Optional[jid.JID],
 options: Optional[dict],
-subscription: pubsub.Subscription
+subscription: pubsub.Subscription,
 ) -> None:
 pass
 async def _unsubscribe_trigger(
 self,
 client = self.host.get_client(profile_key)
 service = client.jid.userhostJID() if not service else jid.JID(service)
 return defer.ensureDeferred(self.synchronise(client, service, node))
 async def synchronise(
-self,
+self, client: SatXMPPEntity, service: jid.JID, node: str, resync: bool = True
-client: SatXMPPEntity,
-service: jid.JID,
-node: str,
-resync: bool = True
 ) -> None:
 """Synchronise a node with a pubsub service
 The node will be synchronised even if there is no matching analyser.
 pubsub_node = await self.host.memory.storage.get_pubsub_node(
 client, service, node
 )
 if pubsub_node is None:
 log.info(
-_(
+_("Synchronising the new node {node} at {service}").format(
-"Synchronising the new node {node} at {service}"
+node=node, service=service.full
-).format(node=node, service=service.full)
+)
 )
 analyse = await self.analyse_node(client, service, node)
 pubsub_node = await self.host.memory.storage.set_pubsub_node(
 client,
 service,
 node,
 analyser=analyse.get("name"),
 type_=analyse.get("type"),
 )
 elif not resync and pubsub_node.sync_state is not None:
 # the node exists, nothing to do
 return
-if ((pubsub_node.sync_state == SyncState.IN_PROGRESS
+if (
-or (service, node) in self.in_progress)):
+pubsub_node.sync_state == SyncState.IN_PROGRESS
+or (service, node) in self.in_progress
+):
 log.warning(
 _(
 "{node} at {service} is already being synchronised, can't do a new "
 "synchronisation."
 ).format(node=node, service=service)

Mercurial > libervia-backend

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