libervia-backend: libervia/backend/plugins/plugin_xep

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

Reformatted code base using black.

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

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 # rsm_request is the rsm.RSMRequest build with rsm_ prefixed keys, or None
 # extra is a potentially empty dict
 TIMEOUT = 30
 # minimum features that a pubsub service must have to be selectable as default
 DEFAULT_PUBSUB_MIN_FEAT = {
-'http://jabber.org/protocol/pubsub#persistent-items',
+"http://jabber.org/protocol/pubsub#persistent-items",
-'http://jabber.org/protocol/pubsub#publish',
+"http://jabber.org/protocol/pubsub#publish",
-'http://jabber.org/protocol/pubsub#retract-items',
+"http://jabber.org/protocol/pubsub#retract-items",
 }
 class XEP_0060(object):
 OPT_ACCESS_MODEL = "pubsub#access_model"
 OPT_PERSIST_ITEMS = "pubsub#persist_items"
 OPT_MAX_ITEMS = "pubsub#max_items"
 def __init__(self, host):
 log.info(_("PubSub plugin initialization"))
 self.host = host
 self._rsm = host.plugins.get("XEP-0059")
 self._mam = host.plugins.get("XEP-0313")
-self._node_cb = {}  # dictionnary of callbacks for node (key: node, value: list of callbacks)
+self._node_cb = (
+{}
+)  # dictionnary of callbacks for node (key: node, value: list of callbacks)
 self.rt_sessions = sat_defer.RTDeferredSessions()
 host.bridge.add_method(
 "ps_node_create",
 ".plugin",
 in_sign="ssa{ss}s",
 client = self.host.get_client(profile)
 except exceptions.ProfileNotSetError:
 return {}
 try:
 return {
-"service": client.pubsub_service.full()
+"service": (
-if client.pubsub_service is not None
+client.pubsub_service.full()
-else ""
+if client.pubsub_service is not None
+else ""
+)
 }
 except AttributeError:
 if self.host.is_connected(profile):
 log.debug("Profile is not connected, service is not checked yet")
 else:
 assert "mam" not in extra
 extra["mam"] = mam_request
 return Extra(rsm_request, extra)
-def add_managed_node(
+def add_managed_node(self, node: str, priority: int = 0, **kwargs: Callable):
-self,
-node: str,
-priority: int = 0,
-**kwargs: Callable
-):
 """Add a handler for a node
 @param node: node to monitor
 all node *prefixed* with this one will be triggered
 @param priority: priority of the callback. Callbacks with higher priority will be
 #     """
 #     d = self.subscriptions(service, nodeIdentifier, profile_key=profile)
 #     d.addCallback(lambda subs: [sub.getAttribute('node') for sub in subs if sub.getAttribute('subscription') == filter_])
 #     return d
-def _send_item(self, service, nodeIdentifier, payload, item_id=None, extra_ser="",
+def _send_item(
-profile_key=C.PROF_KEY_NONE):
+self,
+service,
+nodeIdentifier,
+payload,
+item_id=None,
+extra_ser="",
+profile_key=C.PROF_KEY_NONE,
+):
 client = self.host.get_client(profile_key)
 service = None if not service else jid.JID(service)
 payload = xml_tools.parse(payload)
 extra = data_format.deserialise(extra_ser)
-d = defer.ensureDeferred(self.send_item(
+d = defer.ensureDeferred(
-client, service, nodeIdentifier, payload, item_id or None, extra
+self.send_item(
-))
+client, service, nodeIdentifier, payload, item_id or None, extra
+)
+)
 d.addCallback(lambda ret: ret or "")
 return d
-def _send_items(self, service, nodeIdentifier, items, extra_ser=None,
+def _send_items(
-profile_key=C.PROF_KEY_NONE):
+self, service, nodeIdentifier, items, extra_ser=None, profile_key=C.PROF_KEY_NONE
+):
 client = self.host.get_client(profile_key)
 service = None if not service else jid.JID(service)
 try:
 items = [xml_tools.parse(item) for item in items]
 except Exception as e:
-raise exceptions.DataError(_("Can't parse items: {msg}").format(
+raise exceptions.DataError(_("Can't parse items: {msg}").format(msg=e))
-msg=e))
 extra = data_format.deserialise(extra_ser)
-return defer.ensureDeferred(self.send_items(
+return defer.ensureDeferred(
-client, service, nodeIdentifier, items, extra=extra
+self.send_items(client, service, nodeIdentifier, items, extra=extra)
-))
+)
 async def send_item(
 self,
 client: SatXMPPClient,
 service: Union[jid.JID, None],
 nodeIdentifier: str,
 payload: domish.Element,
 item_id: Optional[str] = None,
-extra: Optional[Dict[str, Any]] = None
+extra: Optional[Dict[str, Any]] = None,
 ) -> Optional[str]:
 """High level method to send one item
 @param service: service to send the item to None to use PEP
 @param NodeIdentifier: PubSub node to use
 @param item_id: id to use or None to create one
 @param extra: extra options
 @return: id of the created item
 """
 assert isinstance(payload, domish.Element)
-item_elt = domish.Element((pubsub.NS_PUBSUB, 'item'))
+item_elt = domish.Element((pubsub.NS_PUBSUB, "item"))
 if item_id is not None:
-item_elt['id'] = item_id
+item_elt["id"] = item_id
 item_elt.addChild(payload)
 published_ids = await self.send_items(
-client,
+client, service, nodeIdentifier, [item_elt], extra=extra
-service,
-nodeIdentifier,
-[item_elt],
-extra=extra
 )
 try:
 return published_ids[0]
 except IndexError:
 return item_id
 client: SatXMPPEntity,
 service: Optional[jid.JID],
 nodeIdentifier: str,
 items: List[domish.Element],
 sender: Optional[jid.JID] = None,
-extra: Optional[Dict[str, Any]] = None
+extra: Optional[Dict[str, Any]] = None,
 ) -> List[str]:
 """High level method to send several items at once
 @param service: service to send the item to
 None to use PEP
 extra = {}
 if service is None:
 service = client.jid.userhostJID()
 parsed_items = []
 for item in items:
-if item.name != 'item':
+if item.name != "item":
 raise exceptions.DataError(_("Invalid item: {xml}").format(item.toXml()))
 item_id = item.getAttribute("id")
 parsed_items.append(pubsub.Item(id=item_id, payload=item.firstChildElement()))
 publish_options = extra.get(self.EXTRA_PUBLISH_OPTIONS)
 try:
 iq_result = await self.publish(
-client, service, nodeIdentifier, parsed_items, options=publish_options,
+client,
-sender=sender
+service,
+nodeIdentifier,
+parsed_items,
+options=publish_options,
+sender=sender,
 )
 except error.StanzaError as e:
-if ((e.condition == 'conflict' and e.appCondition
+if (
-and e.appCondition.name == 'precondition-not-met'
+e.condition == "conflict"
-and publish_options is not None)):
+and e.appCondition
+and e.appCondition.name == "precondition-not-met"
+and publish_options is not None
+):
 # this usually happens when publish-options can't be set
-policy = extra.get(self.EXTRA_ON_PRECOND_NOT_MET, 'raise')
+policy = extra.get(self.EXTRA_ON_PRECOND_NOT_MET, "raise")
-if policy == 'raise':
+if policy == "raise":
 raise e
-elif policy == 'publish_without_options':
+elif policy == "publish_without_options":
-log.warning(_(
+log.warning(
-"Can't use publish-options ({options}) on node {node}, "
+_(
-"re-publishing without them: {reason}").format(
+"Can't use publish-options ({options}) on node {node}, "
-options=', '.join(f'{k} = {v}'
+"re-publishing without them: {reason}"
-for k,v in publish_options.items()),
+).format(
+options=", ".join(
+f"{k} = {v}" for k, v in publish_options.items()
+),
 node=nodeIdentifier,
 reason=e,
 )
 )
 iq_result = await self.publish(
-client, service, nodeIdentifier, parsed_items)
+client, service, nodeIdentifier, parsed_items
+)
 else:
 raise exceptions.InternalError(
 f"Invalid policy in extra's {self.EXTRA_ON_PRECOND_NOT_MET!r}: "
 f"{policy}"
 )
 else:
 raise e
 try:
 return [
-item['id']
+item["id"]
-for item in iq_result.pubsub.publish.elements(pubsub.NS_PUBSUB, 'item')
+for item in iq_result.pubsub.publish.elements(pubsub.NS_PUBSUB, "item")
 ]
 except AttributeError:
 return []
 async def publish(
 service: jid.JID,
 nodeIdentifier: str,
 items: Optional[List[domish.Element]] = None,
 options: Optional[dict] = None,
 sender: Optional[jid.JID] = None,
-extra: Optional[Dict[str, Any]] = None
+extra: Optional[Dict[str, Any]] = None,
 ) -> domish.Element:
 """Publish pubsub items
 @param sender: sender of the request,
 client.jid will be used if nto set
 if sender is None:
 sender = client.jid
 if extra is None:
 extra = {}
 if not await self.host.trigger.async_point(
-"XEP-0060_publish", client, service, nodeIdentifier, items, options, sender,
+"XEP-0060_publish",
-extra
+client,
+service,
+nodeIdentifier,
+items,
+options,
+sender,
+extra,
 ):
 return extra["iq_result_elt"]
 iq_result_elt = await client.pubsub_client.publish(
-service, nodeIdentifier, items, sender,
+service, nodeIdentifier, items, sender, options=options
-options=options
 )
 return iq_result_elt
 def _unwrap_mam_message(self, message_elt):
 try:
 item_elt = reduce(
 lambda elt, ns_name: next(elt.elements(*ns_name)),
-(message_elt,
+(
-(mam.NS_MAM, "result"),
+message_elt,
-(C.NS_FORWARD, "forwarded"),
+(mam.NS_MAM, "result"),
-(C.NS_CLIENT, "message"),
+(C.NS_FORWARD, "forwarded"),
-("http://jabber.org/protocol/pubsub#event", "event"),
+(C.NS_CLIENT, "message"),
-("http://jabber.org/protocol/pubsub#event", "items"),
+("http://jabber.org/protocol/pubsub#event", "event"),
-("http://jabber.org/protocol/pubsub#event", "item"),
+("http://jabber.org/protocol/pubsub#event", "items"),
-))
+("http://jabber.org/protocol/pubsub#event", "item"),
+),
+)
 except StopIteration:
 raise exceptions.DataError("Can't find Item in MAM message element")
 return item_elt
 def serialise_items(self, items_data):
 items, metadata = items_data
-metadata['items'] = items
+metadata["items"] = items
 return data_format.serialise(metadata)
-def _get_items(self, service="", node="", max_items=10, item_ids=None, sub_id=None,
+def _get_items(
-extra="", profile_key=C.PROF_KEY_NONE):
+self,
+service="",
+node="",
+max_items=10,
+item_ids=None,
+sub_id=None,
+extra="",
+profile_key=C.PROF_KEY_NONE,
+):
 """Get items from pubsub node
 @param max_items(int): maximum number of item to get, C.NO_LIMIT for no limit
 """
 client = self.host.get_client(profile_key)
 service = jid.JID(service) if service else None
 max_items = None if max_items == C.NO_LIMIT else max_items
 extra = self.parse_extra(data_format.deserialise(extra))
-d = defer.ensureDeferred(self.get_items(
+d = defer.ensureDeferred(
-client,
+self.get_items(
-service,
+client,
-node,
+service,
-max_items,
+node,
-item_ids,
+max_items,
-sub_id or None,
+item_ids,
-extra.rsm_request,
+sub_id or None,
-extra.extra,
+extra.rsm_request,
-))
+extra.extra,
+)
+)
 d.addCallback(self.trans_items_data)
 d.addCallback(self.serialise_items)
 return d
 async def get_items(
 node: str,
 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] = None
+extra: Optional[dict] = None,
 ) -> Tuple[List[domish.Element], dict]:
 """Retrieve pubsub items from a node.
 @param service (JID, None): pubsub service.
 @param node (str): node id.
 if rsm_request and item_ids:
 raise ValueError("items_id can't be used with rsm")
 if extra is None:
 extra = {}
 cont, ret = await self.host.trigger.async_return_point(
-"XEP-0060_getItems", client, service, node, max_items, item_ids, sub_id,
+"XEP-0060_getItems",
-rsm_request, extra
+client,
+service,
+node,
+max_items,
+item_ids,
+sub_id,
+rsm_request,
+extra,
 )
 if not cont:
 return ret
 try:
 mam_query = extra["mam"]
 except KeyError:
-d = defer.ensureDeferred(client.pubsub_client.items(
+d = defer.ensureDeferred(
-service = service,
+client.pubsub_client.items(
-nodeIdentifier = node,
+service=service,
-maxItems = max_items,
+nodeIdentifier=node,
-subscriptionIdentifier = sub_id,
+maxItems=max_items,
-sender = None,
+subscriptionIdentifier=sub_id,
-itemIdentifiers = item_ids,
+sender=None,
-orderBy = extra.get(C.KEY_ORDER_BY),
+itemIdentifiers=item_ids,
-rsm_request = rsm_request,
+orderBy=extra.get(C.KEY_ORDER_BY),
-extra = extra
+rsm_request=rsm_request,
-))
+extra=extra,
+)
+)
 # we have no MAM data here, so we add None
 d.addErrback(sat_defer.stanza_2_not_found)
 d.addTimeout(TIMEOUT, reactor)
 items, rsm_response = await d
 mam_response = None
 if mam_response is not None:
 # mam_response is a dict with "complete" and "stable" keys
 # we can put them directly in metadata
 metadata.update(mam_response)
 if rsm_request is not None and rsm_response is not None:
-metadata['rsm'] = rsm_response.toDict()
+metadata["rsm"] = rsm_response.toDict()
 if mam_response is None:
 index = rsm_response.index
 count = rsm_response.count
 if index is None or count is None:
 # we don't have enough information to know if the data is complete
 #             log.debug(u"Skip the items retrieval for [{node}]: node doesn't exist".format(node=node))
 #             continue  # avoid pubsub "item-not-found" error
 #         d_dict[publisher] = self.get_items(service, node, max_items, None, sub_id, rsm, client.profile)
 #     defer.returnValue(d_dict)
-def getOptions(self, service, nodeIdentifier, subscriber, subscriptionIdentifier=None,
+def getOptions(
-profile_key=C.PROF_KEY_NONE):
+self,
+service,
+nodeIdentifier,
+subscriber,
+subscriptionIdentifier=None,
+profile_key=C.PROF_KEY_NONE,
+):
 client = self.host.get_client(profile_key)
 return client.pubsub_client.getOptions(
 service, nodeIdentifier, subscriber, subscriptionIdentifier
 )
-def setOptions(self, service, nodeIdentifier, subscriber, options,
+def setOptions(
-subscriptionIdentifier=None, profile_key=C.PROF_KEY_NONE):
+self,
+service,
+nodeIdentifier,
+subscriber,
+options,
+subscriptionIdentifier=None,
+profile_key=C.PROF_KEY_NONE,
+):
 client = self.host.get_client(profile_key)
 return client.pubsub_client.setOptions(
 service, nodeIdentifier, subscriber, options, subscriptionIdentifier
 )
 def createNode(
 self,
 client: SatXMPPClient,
 service: jid.JID,
 nodeIdentifier: Optional[str] = None,
-options: Optional[Dict[str, str]] = None
+options: Optional[Dict[str, str]] = None,
 ) -> str:
 """Create a new node
 @param service: PubSub service,
 @param NodeIdentifier: node name use None to create instant node (identifier will
 def _get_node_affiliations(self, service_s, nodeIdentifier, profile_key):
 client = self.host.get_client(profile_key)
 d = self.get_node_affiliations(
 client, jid.JID(service_s) if service_s else None, nodeIdentifier
 )
-d.addCallback(
+d.addCallback(lambda affiliations: {j.full(): a for j, a in affiliations.items()})
-lambda affiliations: {j.full(): a for j, a in affiliations.items()}
-)
 return d
 def get_node_affiliations(self, client, service, nodeIdentifier):
 """Retrieve affiliations of a node owned by profile"""
 request = pubsub.PubSubRequest("affiliationsGet")
 return self.deleteNode(
 client, jid.JID(service_s) if service_s else None, nodeIdentifier
 )
 def deleteNode(
-self,
+self, client: SatXMPPClient, service: jid.JID, nodeIdentifier: str
-client: SatXMPPClient,
-service: jid.JID,
-nodeIdentifier: str
 ) -> defer.Deferred:
 return client.pubsub_client.deleteNode(service, nodeIdentifier)
 def _addWatch(self, service_s, node, profile_key):
 """watch modifications on a node
 new_id,
 profile_key=C.PROF_KEY_NONE,
 ):
 client = self.host.get_client(profile_key)
 service = jid.JID(service) if service else None
-return defer.ensureDeferred(self.rename_item(
+return defer.ensureDeferred(
-client, service, node, item_id, new_id
+self.rename_item(client, service, node, item_id, new_id)
-))
+)
 async def rename_item(
 self,
 client: SatXMPPEntity,
 service: Optional[jid.JID],
 node: str,
 item_id: str,
-new_id: str
+new_id: str,
 ) -> None:
 """Rename an item by recreating it then deleting it
 we have to recreate then delete because there is currently no rename operation
 with PubSub
 def _subscribe(self, service, nodeIdentifier, options, profile_key=C.PROF_KEY_NONE):
 client = self.host.get_client(profile_key)
 service = None if not service else jid.JID(service)
 d = defer.ensureDeferred(
 self.subscribe(
-client,
+client, service, nodeIdentifier, options=data_format.deserialise(options)
-service,
-nodeIdentifier,
-options=data_format.deserialise(options)
 )
 )
 d.addCallback(lambda subscription: subscription.subscriptionIdentifier or "")
 return d
 self,
 client: SatXMPPEntity,
 service: Optional[jid.JID],
 nodeIdentifier: str,
 sub_jid: Optional[jid.JID] = None,
-options: Optional[dict] = None
+options: Optional[dict] = None,
 ) -> pubsub.Subscription:
 # TODO: reimplement a subscribtion cache, checking that we have not subscription before trying to subscribe
 if service is None:
 service = client.jid.userhostJID()
 cont, trigger_sub = await self.host.trigger.async_return_point(
-"XEP-0060_subscribe", client, service, nodeIdentifier, sub_jid, options,
+"XEP-0060_subscribe",
+client,
+service,
+nodeIdentifier,
+sub_jid,
+options,
 )
 if not cont:
 return trigger_sub
 try:
 subscription = await client.pubsub_client.subscribe(
-service, nodeIdentifier, sub_jid or client.jid.userhostJID(),
+service,
-options=options, sender=client.jid.userhostJID()
+nodeIdentifier,
+sub_jid or client.jid.userhostJID(),
+options=options,
+sender=client.jid.userhostJID(),
 )
 except error.StanzaError as e:
-if e.condition == 'item-not-found':
+if e.condition == "item-not-found":
 raise exceptions.NotFound(e.text or e.condition)
 else:
 raise e
 return subscription
 sub_jid: Optional[jid.JID] = None,
 subscriptionIdentifier: Optional[str] = None,
 sender: Optional[jid.JID] = None,
 ) -> None:
 if not await self.host.trigger.async_point(
-"XEP-0060_unsubscribe", client, service, nodeIdentifier, sub_jid,
+"XEP-0060_unsubscribe",
-subscriptionIdentifier, sender
+client,
+service,
+nodeIdentifier,
+sub_jid,
+subscriptionIdentifier,
+sender,
 ):
 return
 try:
 await client.pubsub_client.unsubscribe(
 service,
 nodeIdentifier,
 sub_jid or client.jid.userhostJID(),
 subscriptionIdentifier,
 sender,
 )
 except error.StanzaError as e:
 try:
 next(e.getElement().elements(pubsub.NS_PUBSUB_ERRORS, "not-subscribed"))
 except StopIteration:
 raise e
 f"subscribed to node {nodeIdentifier!s} at {service.full()}"
 )
 @utils.ensure_deferred
 async def _subscriptions(
-self,
+self, service="", nodeIdentifier="", profile_key=C.PROF_KEY_NONE
-service="",
-nodeIdentifier="",
-profile_key=C.PROF_KEY_NONE
 ) -> str:
 client = self.host.get_client(profile_key)
 service = None if not service else jid.JID(service)
 subs = await self.subscriptions(client, service, nodeIdentifier or None)
 return data_format.serialise(subs)
 async def subscriptions(
 self,
 client: SatXMPPEntity,
 service: Optional[jid.JID] = None,
-node: Optional[str] = None
+node: Optional[str] = None,
 ) -> List[Dict[str, Union[str, bool]]]:
 """Retrieve subscriptions from a service
 @param service(jid.JID): PubSub service
 @param nodeIdentifier(unicode, None): node to check
 def eb(failure_):
 log.warning(f"Error while parsing item: {failure_.value}")
 d = defer.gatherResults([item_cb(item).addErrback(eb) for item in items])
-d.addCallback(lambda parsed_items: (
+d.addCallback(
-[i for i in parsed_items if i is not None],
+lambda parsed_items: ([i for i in parsed_items if i is not None], metadata)
-metadata
+)
-))
 return d
 def ser_d_list(self, results, failure_result=None):
 """Serialise a DeferredList result
 - result
 """
 if failure_result is None:
 failure_result = ()
 return [
-("", result)
+(
-if success
+("", result)
-else (str(result.result) or UNSPECIFIED, failure_result)
+if success
+else (str(result.result) or UNSPECIFIED, failure_result)
+)
 for success, result in results
 ]
 # subscribe #
 @utils.ensure_deferred
 async def _get_node_subscriptions(
-self,
+self, service: str, node: str, profile_key: str
-service: str,
-node: str,
-profile_key: str
 ) -> Dict[str, str]:
 client = self.host.get_client(profile_key)
 subs = await self.get_node_subscriptions(
 client, jid.JID(service) if service else None, node
 )
 return {j.full(): a for j, a in subs.items()}
 async def get_node_subscriptions(
-self,
+self, client: SatXMPPEntity, service: Optional[jid.JID], nodeIdentifier: str
-client: SatXMPPEntity,
-service: Optional[jid.JID],
-nodeIdentifier: str
 ) -> Dict[jid.JID, str]:
 """Retrieve subscriptions to a node
 @param nodeIdentifier(unicode): node to get subscriptions from
 """
 except AttributeError as e:
 raise ValueError(_("Invalid result: {}").format(e))
 try:
 return {
 jid.JID(s["jid"]): s["subscription"]
-for s in subscriptions_elt.elements(
+for s in subscriptions_elt.elements((pubsub.NS_PUBSUB, "subscription"))
-(pubsub.NS_PUBSUB, "subscription")
-)
 }
 except KeyError:
 raise ValueError(
-_("Invalid result: bad <subscription> element: {}").format(
+_("Invalid result: bad <subscription> element: {}").format(iq_elt.toXml)
-iq_elt.toXml
-)
 )
 def _set_node_subscriptions(
 self, service_s, nodeIdentifier, subscriptions, profile_key=C.PROF_KEY_NONE
 ):
 client = self.host.get_client(profile_key)
 subscriptions = {
-jid.JID(jid_): subscription
+jid.JID(jid_): subscription for jid_, subscription in subscriptions.items()
-for jid_, subscription in subscriptions.items()
 }
 d = self.set_node_subscriptions(
 client,
 jid.JID(service_s) if service_s else None,
 nodeIdentifier,
 """
 client = self.host.get_client(profile_key)
 deferreds = {}
 for service, node in node_data:
 deferreds[(service, node)] = defer.ensureDeferred(
-client.pubsub_client.subscribe(
+client.pubsub_client.subscribe(service, node, subscriber, options=options)
-service, node, subscriber, options=options
-)
 )
 return self.rt_sessions.new_session(deferreds, client.profile)
 # found_nodes = yield self.listNodes(service, profile=client.profile)
 # subscribed_nodes = yield self.listSubscribedNodes(service, profile=client.profile)
 # d_list = []
 extra.rsm_request,
 extra.extra,
 profile_key,
 )
-def get_from_many(self, node_data, max_item=None, rsm_request=None, extra=None,
+def get_from_many(
-profile_key=C.PROF_KEY_NONE):
+self,
+node_data,
+max_item=None,
+rsm_request=None,
+extra=None,
+profile_key=C.PROF_KEY_NONE,
+):
 """Get items from many nodes at once
 @param node_data (iterable[tuple]): iterable of tuple (service, node) where:
 - service (jid.JID) is the pubsub service
 - node (unicode) is the node to get items from
 @return (str): RT Deferred session id
 """
 client = self.host.get_client(profile_key)
 deferreds = {}
 for service, node in node_data:
-deferreds[(service, node)] = defer.ensureDeferred(self.get_items(
+deferreds[(service, node)] = defer.ensureDeferred(
-client, service, node, max_item, rsm_request=rsm_request, extra=extra
+self.get_items(
-))
+client, service, node, max_item, rsm_request=rsm_request, extra=extra
+)
+)
 return self.rt_sessions.new_session(deferreds, client.profile)
 @implementer(disco.IDisco)
 class SatPubSubClient(rsm.PubSubClient):
 extra: Optional[Dict[str, Any]] = None,
 ):
 if extra is None:
 extra = {}
 items, rsm_response = await super().items(
-service, nodeIdentifier, maxItems, subscriptionIdentifier, sender,
+service,
-itemIdentifiers, orderBy, rsm_request
+nodeIdentifier,
+maxItems,
+subscriptionIdentifier,
+sender,
+itemIdentifiers,
+orderBy,
+rsm_request,
 )
 # items must be returned, thus this async point can't stop the workflow (but it
 # can modify returned items)
 await self.host.trigger.async_point(
-"XEP-0060_items", self.parent, service, nodeIdentifier, items, rsm_response,
+"XEP-0060_items",
-extra
+self.parent,
+service,
+nodeIdentifier,
+items,
+rsm_response,
+extra,
 )
 return items, rsm_response
 def _get_node_callbacks(self, node, event):
 """Generate callbacks from given node and event
 """
 for callback in self._get_node_callbacks(event.nodeIdentifier, C.PS_ITEMS):
 try:
 await utils.as_deferred(callback, client, event)
 except Exception as e:
-log.error(
+log.error(f"Error while running items event callback {callback}: {e}")
-f"Error while running items event callback {callback}: {e}"
-)
 def itemsReceived(self, event):
 log.debug("Pubsub items received")
 client = self.parent
 defer.ensureDeferred(self._call_node_callbacks(client, event))
 def deleteReceived(self, event):
 log.debug(("Publish node deleted"))
 for callback in self._get_node_callbacks(event.nodeIdentifier, C.PS_DELETE):
 d = utils.as_deferred(callback, self.parent, event)
-d.addErrback(lambda f: log.error(
+d.addErrback(
-f"Error while running delete event callback {callback}: {f}"
+lambda f: log.error(
-))
+f"Error while running delete event callback {callback}: {f}"
+)
+)
 client = self.parent
 if (event.sender, event.nodeIdentifier) in client.pubsub_watching:
 self.host.bridge.ps_event_raw(
 event.sender.full(), event.nodeIdentifier, C.PS_DELETE, [], client.profile
 )
 def purgeReceived(self, event):
 log.debug(("Publish node purged"))
 for callback in self._get_node_callbacks(event.nodeIdentifier, C.PS_PURGE):
 d = utils.as_deferred(callback, self.parent, event)
-d.addErrback(lambda f: log.error(
+d.addErrback(
-f"Error while running purge event callback {callback}: {f}"
+lambda f: log.error(
-))
+f"Error while running purge event callback {callback}: {f}"
+)
+)
 client = self.parent
 if (event.sender, event.nodeIdentifier) in client.pubsub_watching:
 self.host.bridge.ps_event_raw(
 event.sender.full(), event.nodeIdentifier, C.PS_PURGE, [], client.profile
 )
 @param service(jid.JID, None): service to send the item to
 None to use PEP
 @param NodeIdentifier(unicode): PubSub node to use
 """
 # TODO: propose this upstream and remove it once merged
-request = pubsub.PubSubRequest('purge')
+request = pubsub.PubSubRequest("purge")
 request.recipient = service
 request.nodeIdentifier = nodeIdentifier
 return request.send(self.xmlstream)
 def getDiscoInfo(self, requestor, service, nodeIdentifier=""):

Mercurial > libervia-backend

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