libervia-backend: libervia/backend/plugins/plugin_xep

comparison 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

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 "ps_schema_ui_get",
 ".plugin",
 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(
 "ps_schema_dict_get",
 ".plugin",
 return SchemaHandler()
 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:
 """Returns node used for data template (i.e. schema)"""
 app_ns = self.get_application_ns(namespace)
 """
 app_ns = self.get_application_ns(nodeIdentifier)
 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:
 log.warning(
 f"No schema found in item of {service!r} at node {nodeIdentifier!r}: "
 f"\n{items_data[0][0].toXml()}"
 )
 schema = None
 return schema
-async def get_schema_form(self, client, service, nodeIdentifier, schema=None,
+async def get_schema_form(
-form_type="form", copy_form=True):
+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
 @param nodeIdentifier(unicode): node
 @param schema(domish.Element, data_form.Form, None): node schema
 if copy_form:
 # XXX: we don't use deepcopy as it will do an infinite loop if a
 #      domish.Element is present in the form fields (happens for
 #      XEP-0315 data forms XML Element)
 schema = data_form.Form(
-formType = schema.formType,
+formType=schema.formType,
-title = schema.title,
+title=schema.title,
-instructions = schema.instructions[:],
+instructions=schema.instructions[:],
-formNamespace = schema.formNamespace,
+formNamespace=schema.formNamespace,
-fields = schema.fieldList,
+fields=schema.fieldList,
 )
 return schema
 try:
 form = data_form.Form.fromElement(schema)
 def schema_2_xmlui(self, schema_elt):
 form = data_form.Form.fromElement(schema_elt)
 xmlui = xml_tools.data_form_2_xmlui(form, "")
 return xmlui
-def _get_ui_schema(self, service, nodeIdentifier, default_node=None,
+def _get_ui_schema(
-profile_key=C.PROF_KEY_NONE):
+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"))
 nodeIdentifier = default_node
 client = self.host.get_client(profile_key)
 d = defer.ensureDeferred(self.get_schema_dict(client, service, nodeIdentifier))
 d.addCallback(data_format.serialise)
 return d
 async def get_schema_dict(
-self,
+self, client: SatXMPPEntity, service: Optional[jid.JID], nodeIdentifier: str
-client: SatXMPPEntity,
+) -> dict:
-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
 """
 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,
+def _get_data_form_items(
-item_ids=None, sub_id=None, extra="",
+self,
-profile_key=C.PROF_KEY_NONE):
+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:
 raise exceptions.DataError(_("empty node is not allowed"))
 if schema:
 )
 )
 d.addCallback(self._p.trans_items_data)
 return d
-async def get_data_form_items(self, client, service, nodeIdentifier, schema=None,
+async def get_data_form_items(
-max_items=None, item_ids=None, sub_id=None, rsm_request=None,
+self,
-extra=None, default_node=None, form_ns=None, filters=None):
+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
 if None, it will be retrieved from node
 @param default_node(unicode): node to use if nodeIdentifier is None or empty
 ("label", "id"),
 ("text", item_elt["id"], "id"),
 ("label", "publisher"),
 ]
 try:
-publisher = jid.JID(item_elt['publisher'])
+publisher = jid.JID(item_elt["publisher"])
 except (KeyError, jid.InvalidFormat):
 pass
 else:
 prepend.append(("jid", publisher, "publisher"))
 xmlui = xml_tools.data_form_result_2_xmlui(
 )
 items_xmlui.append(xmlui)
 break
 return (items_xmlui, metadata)
-def _send_data_form_item(self, service, nodeIdentifier, values, schema=None,
+def _send_data_form_item(
-item_id=None, extra=None, profile_key=C.PROF_KEY_NONE):
+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:
 schema = generic.parseXml(schema.encode("utf-8"))
 else:
 )
 d.addCallback(lambda ret: ret or "")
 return d
 async def send_data_form_item(
-self, client, service, nodeIdentifier, values, schema=None, item_id=None,
+self,
-extra=None, deserialise=False):
+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
 form. If not iterable, will be put in a list.
 @param schema(domish.Element, data_form.Form, None): data schema
 except KeyError:
 log.warning(
 _("field {name} doesn't exist, ignoring it").format(name=name)
 )
 continue
-if isinstance(values_list, str) or not isinstance(
+if isinstance(values_list, str) or not isinstance(values_list, Iterable):
-values_list, Iterable
-):
 values_list = [values_list]
 if deserialise:
 if field.fieldType == "boolean":
 values_list = [C.bool(v) for v in values_list]
 elif field.fieldType == "text-multi":
 # for text-multi, lines must be put on separate values
 values_list = list(
 itertools.chain(*[v.splitlines() for v in values_list])
 )
 elif xml_tools.is_xhtml_field(field):
-values_list = [generic.parseXml(v.encode("utf-8"))
+values_list = [
-for v in 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 ""):
 # for lists, we check that given values are allowed in form
 allowed_values = [o.value for o in field.options]
 values_list = [v for v in values_list if v in allowed_values]
 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
+if field.value and (
-or value.uri != field.value.uri)):
+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:
 # only XHTML fields are handled for now
 raise NotImplementedError
 sub_id = None
 extra = self._p.parse_extra(extra)
 return client, service, node, max_items, extra, sub_id
-def _get(self, service="", node="", max_items=10, item_ids=None, sub_id=None,
+def _get(
-extra="", default_node=None, form_ns=None, filters=None,
+self,
-profile_key=C.PROF_KEY_NONE):
+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
 when adding *Get methods
 extra can have the key "labels_as_list" which is a hack to convert
 client, service, node, item_ids=[item_id]
 )
 item_elt = items_data[0][0]
 except Exception as e:
 log.warning(
-_("Can't get previous item, update ignored: {reason}").format(
+_("Can't get previous item, update ignored: {reason}").format(reason=e)
-reason=e
-)
 )
 else:
 # and parse it
 form = data_form.findForm(item_elt, form_ns)
 if form is None:
 else:
 for name, field in form.fields.items():
 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,
+def _set(
-default_node=None, form_ns=None, fill_author=True,
+self,
-profile_key=C.PROF_KEY_NONE):
+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
 when adding *Set methods
 """
 client, service, node, schema, item_id, extra = self.prepare_bridge_set(
 service, node, schema, item_id, extra
 )
-d = defer.ensureDeferred(self.set(
+d = defer.ensureDeferred(
-client,
+self.set(
-service,
+client,
-node,
+service,
-values,
+node,
-schema,
+values,
-item_id,
+schema,
-extra,
+item_id,
-deserialise=True,
+extra,
-form_ns=form_ns,
+deserialise=True,
-default_node=default_node,
+form_ns=form_ns,
-fill_author=fill_author,
+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,
+self,
-form_ns, default_node=None, fill_author=True):
+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
 @param values(dict[key(unicode), [iterable[object]|object]]): values of the items
 if value is not iterable, it will be put in a list
 elif extra.get("update", False):
 if item_id is None:
 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(
 client, service, node, values, schema, item_id, extra, deserialise
 )

Mercurial > libervia-backend

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