libervia-backend: libervia/backend/plugins/plugin_xep

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

Reformatted code base using black.

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

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 def register_attachment_handler(
 self,
 name: str,
 namespace: str,
-get_cb: Callable[
+get_cb: Callable[[SatXMPPEntity, domish.Element, Dict[str, Any]], None],
-[SatXMPPEntity, domish.Element, Dict[str, Any]],
-None],
 set_cb: Callable[
 [SatXMPPEntity, Dict[str, Any], Optional[domish.Element]],
-Optional[domish.Element]],
+Optional[domish.Element],
+],
 ) -> None:
 """Register callbacks to handle an attachment
 @param name: name of the element
 @param namespace: namespace of the element
 key = (name, namespace)
 if key in self.handlers:
 raise exceptions.ConflictError(
 f"({name}, {namespace}) attachment handlers are already registered"
 )
-self.handlers[(name, namespace)] = {
+self.handlers[(name, namespace)] = {"get": get_cb, "set": set_cb}
-"get": get_cb,
-"set": set_cb
-}
 def get_attachment_node_name(self, service: jid.JID, node: str, item: str) -> str:
 """Generate name to use for attachment node"""
 target_item_uri = uri.build_xmpp_uri(
-"pubsub",
+"pubsub", path=service.userhost(), node=node, item=item
-path=service.userhost(),
-node=node,
-item=item
 )
 return f"{NS_PUBSUB_ATTACHMENTS}/{target_item_uri}"
 def is_attachment_node(self, node: str) -> bool:
 """Return True if node name is an attachment node"""
 self,
 client: SatXMPPEntity,
 service: jid.JID,
 node: str,
 item: domish.Element,
-data: dict
+data: dict,
 ) -> bool:
 """trigger to create attachment node on each publication"""
 await self.create_attachments_node(
 client, service, node, item["id"], autocreate=True
 )
 self,
 client: SatXMPPEntity,
 service: jid.JID,
 node: str,
 item_id: str,
-autocreate: bool = False
+autocreate: bool = False,
 ):
 """Create node for attachements if necessary
 @param service: service of target node
 @param node: node where target item is published
 try:
 node_config = await self._p.getConfiguration(client, service, node)
 except error.StanzaError as e:
 if e.condition == "item-not-found" and autocreate:
 # we auto-create the missing node
-await self._p.createNode(
+await self._p.createNode(client, service, node)
-client, service, node
-)
 node_config = await self._p.getConfiguration(client, service, node)
 elif e.condition in ("forbidden", "feature-not-implemented"):
 node_config = self._p.make_configuration_form({})
 else:
 raise e
 )
 except Exception as e:
 log.warning(f"Can't create attachment node {attachment_node}: {e}")
 def items_2_attachment_data(
-self,
+self, client: SatXMPPEntity, items: List[domish.Element]
-client: SatXMPPEntity,
-items: List[domish.Element]
 ) -> List[Dict[str, Any]]:
 """Convert items from attachment node to attachment data"""
 list_data = []
 for item in items:
 try:
 log.warning(
 "item ID is not a JID, this is not compliant and is ignored: "
 f"{item.toXml}"
 )
 continue
-data = {
+data = {"from": item_id}
-"from": item_id
-}
 for handler in self.handlers.values():
 handler["get"](client, attachments_elt, data)
 if len(data) > 1:
 list_data.append(data)
 return list_data
 service_s: str,
 node: str,
 item: str,
 senders_s: List[str],
 extra_s: str,
-profile_key: str
+profile_key: str,
 ) -> defer.Deferred:
 client = self.host.get_client(profile_key)
 extra = data_format.deserialise(extra_s)
 senders = [jid.JID(s) for s in senders_s]
 d = defer.ensureDeferred(
 self.get_attachments(client, jid.JID(service_s), node, item, senders)
 )
 d.addCallback(
-lambda ret:
+lambda ret: (data_format.serialise(ret[0]), data_format.serialise(ret[1]))
-(data_format.serialise(ret[0]),
-data_format.serialise(ret[1]))
 )
 return d
 async def get_attachments(
 self,
 client: SatXMPPEntity,
 service: jid.JID,
 node: str,
 item: str,
 senders: Optional[List[jid.JID]],
-extra: Optional[dict] = None
+extra: Optional[dict] = None,
 ) -> Tuple[List[Dict[str, Any]], dict]:
 """Retrieve data attached to a pubsub item
 @param service: pubsub service where the node is
 @param node: pubsub node containing the item
 )
 list_data = self.items_2_attachment_data(client, items)
 return list_data, metadata
-def _set(
+def _set(self, attachments_s: str, profile_key: str) -> None:
-self,
-attachments_s: str,
-profile_key: str
-) -> None:
 client = self.host.get_client(profile_key)
-attachments = data_format.deserialise(attachments_s)  or {}
+attachments = data_format.deserialise(attachments_s) or {}
 return defer.ensureDeferred(self.set_attachements(client, attachments))
 async def apply_set_handler(
 self,
 client: SatXMPPEntity,
 if new_elt is not None:
 attachments_elt.addChild(new_elt)
 return item_elt
 async def set_attachements(
-self,
+self, client: SatXMPPEntity, attachments_data: Dict[str, Any]
-client: SatXMPPEntity,
-attachments_data: Dict[str, Any]
 ) -> None:
 """Set or update attachments
 Former <attachments> element will be retrieved and updated. Individual
 attachments replace or update their elements individually, according to the
 try:
 service = jid.JID(attachments_data["service"])
 node = attachments_data["node"]
 item = attachments_data["id"]
 except (KeyError, RuntimeError):
-raise ValueError(
+raise ValueError('data must have "service", "node" and "id" set')
-'data must have "service", "node" and "id" set'
-)
 attachment_node = self.get_attachment_node_name(service, node, item)
 try:
 items, __ = await self._p.get_items(
 client, service, attachment_node, item_ids=[client.jid.userhost()]
 )
 @param item: name of target item (used to get attachment node's name)
 """
 attachment_node = self.get_attachment_node_name(service, node, item)
 await self._p.subscribe(client, service, attachment_node)
 def set_timestamp(self, attachment_elt: domish.Element, data: dict) -> None:
 """Check if a ``timestamp`` attribute is set, parse it, and fill data
 @param attachments_elt: element where the ``timestamp`` attribute may be set
 @param data: data specific to the attachment (i.e. not the whole microblog data)
 client: SatXMPPEntity,
 attachments_elt: domish.Element,
 data: Dict[str, Any],
 ) -> None:
 try:
-noticed_elt = next(
+noticed_elt = next(attachments_elt.elements(NS_PUBSUB_ATTACHMENTS, "noticed"))
-attachments_elt.elements(NS_PUBSUB_ATTACHMENTS, "noticed")
-)
 except StopIteration:
 pass
 else:
-noticed_data = {
+noticed_data = {"noticed": True}
-"noticed": True
-}
 self.set_timestamp(noticed_elt, noticed_data)
 data["noticed"] = noticed_data
 def noticed_set(
 self,
 client: SatXMPPEntity,
 data: Dict[str, Any],
-former_elt: Optional[domish.Element]
+former_elt: Optional[domish.Element],
 ) -> Optional[domish.Element]:
 """add or remove a <noticed> attachment
 if data["noticed"] is True, element is added, if it's False, it's removed, and
 it's not present or None, the former element is kept.
 noticed = data["extra"].get("noticed")
 if noticed is None:
 return former_elt
 elif noticed:
 return domish.Element(
-(NS_PUBSUB_ATTACHMENTS, "noticed"),
+(NS_PUBSUB_ATTACHMENTS, "noticed"), attribs={"timestamp": xmpp_date()}
-attribs = {
-"timestamp": xmpp_date()
-}
 )
 else:
 return None
 def reactions_get(
 def reactions_set(
 self,
 client: SatXMPPEntity,
 data: Dict[str, Any],
-former_elt: Optional[domish.Element]
+former_elt: Optional[domish.Element],
 ) -> Optional[domish.Element]:
 """update the <reaction> attachment"""
 reactions_data = data["extra"].get("reactions")
 if reactions_data is None:
 return former_elt
 operation_type = reactions_data.get("operation", "update")
 if operation_type == "update":
-former_reactions = {
+former_reactions = (
-str(r) for r in former_elt.elements(NS_PUBSUB_ATTACHMENTS, "reaction")
+{str(r) for r in former_elt.elements(NS_PUBSUB_ATTACHMENTS, "reaction")}
-} if former_elt is not None else set()
+if former_elt is not None
+else set()
+)
 added_reactions = set(reactions_data.get("add") or [])
 removed_reactions = set(reactions_data.get("remove") or [])
 reactions = list((former_reactions | added_reactions) - removed_reactions)
 elif operation_type == "replace":
 reactions = reactions_data.get("reactions") or []
 else:
 raise exceptions.DataError(f"invalid reaction operation: {operation_type!r}")
 if reactions:
 reactions_elt = domish.Element(
-(NS_PUBSUB_ATTACHMENTS, "reactions"),
+(NS_PUBSUB_ATTACHMENTS, "reactions"), attribs={"timestamp": xmpp_date()}
-attribs = {
-"timestamp": xmpp_date()
-}
 )
 for reactions_data in reactions:
 reactions_elt.addElement("reaction", content=reactions_data)
 return reactions_elt
 else:

Mercurial > libervia-backend

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