Mercurial > libervia-backend
diff libervia/backend/plugins/plugin_xep_0346.py @ 4270:0d7bb4df2343
Reformatted code base using black.
author | Goffi <goffi@goffi.org> |
---|---|
date | Wed, 19 Jun 2024 18:44:57 +0200 |
parents | 4b842c1fb686 |
children |
line wrap: on
line diff
--- a/libervia/backend/plugins/plugin_xep_0346.py Tue Jun 18 12:06:45 2024 +0200 +++ b/libervia/backend/plugins/plugin_xep_0346.py Wed Jun 19 18:44:57 2024 +0200 @@ -83,7 +83,8 @@ in_sign="sss", out_sign="s", method=lambda service, nodeIdentifier, profile_key: self._get_ui_schema( - service, nodeIdentifier, default_node=None, profile_key=profile_key), + service, nodeIdentifier, default_node=None, profile_key=profile_key + ), async_=True, ) host.bridge.add_method( @@ -138,9 +139,9 @@ def get_application_ns(self, namespace): """Retrieve application namespace, i.e. namespace without FDP prefix""" if namespace.startswith(SUBMITTED_PREFIX): - namespace = namespace[len(SUBMITTED_PREFIX):] + namespace = namespace[len(SUBMITTED_PREFIX) :] elif namespace.startswith(TEMPLATE_PREFIX): - namespace = namespace[len(TEMPLATE_PREFIX):] + namespace = namespace[len(TEMPLATE_PREFIX) :] return namespace def get_template_ns(self, namespace: str) -> str: @@ -177,7 +178,7 @@ node_id = f"{TEMPLATE_PREFIX}{app_ns}" items_data = await self._p.get_items(client, service, node_id, max_items=1) try: - schema = next(items_data[0][0].elements(data_form.NS_X_DATA, 'x')) + schema = next(items_data[0][0].elements(data_form.NS_X_DATA, "x")) except IndexError: schema = None except StopIteration: @@ -188,8 +189,15 @@ schema = None return schema - async def get_schema_form(self, client, service, nodeIdentifier, schema=None, - form_type="form", copy_form=True): + async def get_schema_form( + self, + client, + service, + nodeIdentifier, + schema=None, + form_type="form", + copy_form=True, + ): """Get data form from node's schema @param service(None, jid.JID): PubSub service @@ -219,11 +227,11 @@ # domish.Element is present in the form fields (happens for # XEP-0315 data forms XML Element) schema = data_form.Form( - formType = schema.formType, - title = schema.title, - instructions = schema.instructions[:], - formNamespace = schema.formNamespace, - fields = schema.fieldList, + formType=schema.formType, + title=schema.title, + instructions=schema.instructions[:], + formNamespace=schema.formNamespace, + fields=schema.fieldList, ) return schema @@ -239,8 +247,9 @@ xmlui = xml_tools.data_form_2_xmlui(form, "") return xmlui - def _get_ui_schema(self, service, nodeIdentifier, default_node=None, - profile_key=C.PROF_KEY_NONE): + def _get_ui_schema( + self, service, nodeIdentifier, default_node=None, profile_key=C.PROF_KEY_NONE + ): if not nodeIdentifier: if not default_node: raise ValueError(_("nodeIndentifier needs to be set")) @@ -290,10 +299,8 @@ return d async def get_schema_dict( - self, - client: SatXMPPEntity, - service: Optional[jid.JID], - nodeIdentifier: str) -> dict: + self, client: SatXMPPEntity, service: Optional[jid.JID], nodeIdentifier: str + ) -> dict: """Retrieve a node schema and format it a simple dictionary The dictionary is made so it can be easily serialisable @@ -301,9 +308,18 @@ schema_form = await self.get_schema_form(client, service, nodeIdentifier) return xml_tools.data_form_2_data_dict(schema_form) - def _get_data_form_items(self, form_ns="", service="", node="", schema="", max_items=10, - item_ids=None, sub_id=None, extra="", - profile_key=C.PROF_KEY_NONE): + def _get_data_form_items( + self, + form_ns="", + service="", + node="", + schema="", + max_items=10, + item_ids=None, + sub_id=None, + extra="", + profile_key=C.PROF_KEY_NONE, + ): client = self.host.get_client(profile_key) service = jid.JID(service) if service else None if not node: @@ -331,9 +347,21 @@ d.addCallback(self._p.trans_items_data) return d - async def get_data_form_items(self, client, service, nodeIdentifier, schema=None, - max_items=None, item_ids=None, sub_id=None, rsm_request=None, - extra=None, default_node=None, form_ns=None, filters=None): + async def get_data_form_items( + self, + client, + service, + nodeIdentifier, + schema=None, + max_items=None, + item_ids=None, + sub_id=None, + rsm_request=None, + extra=None, + default_node=None, + form_ns=None, + filters=None, + ): """Get items known as being data forms, and convert them to XMLUI @param schema(domish.Element, data_form.Form, None): schema of the node if known @@ -386,7 +414,7 @@ ("label", "publisher"), ] try: - publisher = jid.JID(item_elt['publisher']) + publisher = jid.JID(item_elt["publisher"]) except (KeyError, jid.InvalidFormat): pass else: @@ -404,8 +432,16 @@ break return (items_xmlui, metadata) - def _send_data_form_item(self, service, nodeIdentifier, values, schema=None, - item_id=None, extra=None, profile_key=C.PROF_KEY_NONE): + def _send_data_form_item( + self, + service, + nodeIdentifier, + values, + schema=None, + item_id=None, + extra=None, + profile_key=C.PROF_KEY_NONE, + ): client = self.host.get_client(profile_key) service = None if not service else jid.JID(service) if schema: @@ -428,8 +464,16 @@ return d async def send_data_form_item( - self, client, service, nodeIdentifier, values, schema=None, item_id=None, - extra=None, deserialise=False): + self, + client, + service, + nodeIdentifier, + values, + schema=None, + item_id=None, + extra=None, + deserialise=False, + ): """Publish an item as a dataform when we know that there is a schema @param values(dict[key(unicode), [iterable[object], object]]): values set for the @@ -457,9 +501,7 @@ _("field {name} doesn't exist, ignoring it").format(name=name) ) continue - if isinstance(values_list, str) or not isinstance( - values_list, Iterable - ): + if isinstance(values_list, str) or not isinstance(values_list, Iterable): values_list = [values_list] if deserialise: if field.fieldType == "boolean": @@ -470,8 +512,9 @@ itertools.chain(*[v.splitlines() for v in values_list]) ) elif xml_tools.is_xhtml_field(field): - values_list = [generic.parseXml(v.encode("utf-8")) - for v in values_list] + values_list = [ + generic.parseXml(v.encode("utf-8")) for v in values_list + ] elif "jid" in (field.fieldType or ""): values_list = [jid.JID(v) for v in values_list] if "list" in (field.fieldType or ""): @@ -481,22 +524,25 @@ if not values_list: # if values don't map to allowed values, we use default ones values_list = field.values - elif field.ext_type == 'xml': + elif field.ext_type == "xml": # FIXME: XML elements are not handled correctly, we need to know if we # have actual XML/XHTML, or text to escape for idx, value in enumerate(values_list[:]): if isinstance(value, domish.Element): - if (field.value and (value.name != field.value.name - or value.uri != field.value.uri)): + if field.value and ( + value.name != field.value.name or value.uri != field.value.uri + ): # the element is not the one expected in form, so we create the right element # to wrap the current value - wrapper_elt = domish.Element((field.value.uri, field.value.name)) + wrapper_elt = domish.Element( + (field.value.uri, field.value.name) + ) wrapper_elt.addChild(value) values_list[idx] = wrapper_elt else: # we have to convert the value to a domish.Element if field.value and field.value.uri == C.NS_XHTML: - div_elt = domish.Element((C.NS_XHTML, 'div')) + div_elt = domish.Element((C.NS_XHTML, "div")) div_elt.addContent(str(value)) values_list[idx] = div_elt else: @@ -570,9 +616,19 @@ return client, service, node, max_items, extra, sub_id - def _get(self, service="", node="", max_items=10, item_ids=None, sub_id=None, - extra="", default_node=None, form_ns=None, filters=None, - profile_key=C.PROF_KEY_NONE): + def _get( + self, + service="", + node="", + max_items=10, + item_ids=None, + sub_id=None, + extra="", + default_node=None, + form_ns=None, + filters=None, + profile_key=C.PROF_KEY_NONE, + ): """bridge method to retrieve data from node with schema this method is a helper so dependant plugins can use it directly @@ -647,9 +703,7 @@ item_elt = items_data[0][0] except Exception as e: log.warning( - _("Can't get previous item, update ignored: {reason}").format( - reason=e - ) + _("Can't get previous item, update ignored: {reason}").format(reason=e) ) else: # and parse it @@ -663,9 +717,19 @@ if name not in values: values[name] = "\n".join(str(v) for v in field.values) - def _set(self, service, node, values, schema=None, item_id=None, extra=None, - default_node=None, form_ns=None, fill_author=True, - profile_key=C.PROF_KEY_NONE): + def _set( + self, + service, + node, + values, + schema=None, + item_id=None, + extra=None, + default_node=None, + form_ns=None, + fill_author=True, + profile_key=C.PROF_KEY_NONE, + ): """bridge method to set item in node with schema this method is a helper so dependant plugins can use it directly @@ -674,25 +738,38 @@ client, service, node, schema, item_id, extra = self.prepare_bridge_set( service, node, schema, item_id, extra ) - d = defer.ensureDeferred(self.set( - client, - service, - node, - values, - schema, - item_id, - extra, - deserialise=True, - form_ns=form_ns, - default_node=default_node, - fill_author=fill_author, - )) + d = defer.ensureDeferred( + self.set( + client, + service, + node, + values, + schema, + item_id, + extra, + deserialise=True, + form_ns=form_ns, + default_node=default_node, + fill_author=fill_author, + ) + ) d.addCallback(lambda ret: ret or "") return d async def set( - self, client, service, node, values, schema, item_id, extra, deserialise, - form_ns, default_node=None, fill_author=True): + self, + client, + service, + node, + values, + schema, + item_id, + extra, + deserialise, + form_ns, + default_node=None, + fill_author=True, + ): """Set an item in a node with a schema This method can be used directly by *Set methods added by dependant plugin @@ -725,13 +802,15 @@ raise exceptions.DataError( _('if extra["update"] is set, item_id must be set too') ) - await self.copy_missing_values(client, service, node, item_id, form_ns, values) + await self.copy_missing_values( + client, service, node, item_id, form_ns, values + ) values["updated"] = now if fill_author: if not values.get("author"): id_data = await self._i.get_identity(client, None, ["nicknames"]) - values["author"] = id_data['nicknames'][0] + values["author"] = id_data["nicknames"][0] if not values.get("author_jid"): values["author_jid"] = client.jid.full() item_id = await self.send_data_form_item(