libervia-backend: libervia/backend/core/main.py comparison

comparison libervia/backend/core/main.py @ 4270:0d7bb4df2343

Reformatted code base using black.

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

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 from wokkel.data_form import Option
 from libervia import backend
 from libervia.backend.core.i18n import _, D_, language_switch
 from libervia.backend.core import patches
 patches.apply()
 from twisted.application import service
 from twisted.internet import defer
 from twisted.words.protocols.jabber import jid
 from twisted.internet import reactor
 import libervia.backend.plugins
 log = getLogger(__name__)
 class LiberviaBackend(service.Service):
 def _init(self):
 # we don't use __init__ to avoid doule initialisation with twistd
 # this _init is called in startService
 log.info(f"{C.APP_NAME} {self.full_version}")
 self._cb_map = {}  # map from callback_id to callbacks
 # dynamic menus. key: callback_id, value: menu data (dictionnary)
 self._menus = {}
 self._menus_paths = {}  # path to id. key: (menu_type, lower case tuple of path),
 # value: menu id
 # like initialised, but launched before init script is done, mainly useful for CLI
 # frontend, so it can be used in init script, while other frontends are waiting.
 self.init_pre_script = defer.Deferred()
 self.initialised = defer.Deferred()
 self.profiles = {}
 self.plugins = {}
-# map for short name to whole namespace,
+# map for short name to whole namespace,
 # extended by plugins with register_namespace
 self.ns_map = {
 "x-data": xmpp.NS_X_DATA,
 "disco#info": xmpp.NS_DISCO_INFO,
 }
 self.memory = memory.Memory(self)
 # trigger are used to change Libervia behaviour
-self.trigger = (
+self.trigger = trigger.TriggerManager()
-trigger.TriggerManager()
-)
+bridge_name = os.getenv("LIBERVIA_BRIDGE_NAME") or self.memory.config_get(
+"", "bridge", "dbus"
-bridge_name = (
-os.getenv("LIBERVIA_BRIDGE_NAME")
-or self.memory.config_get("", "bridge", "dbus")
 )
 bridge_module = dynamic_import.bridge(bridge_name)
 if bridge_module is None:
 log.error(f"Can't find bridge module of name {bridge_name}")
 )
 shutil.rmtree(self.local_shared_path, ignore_errors=True)
 self.local_shared_path.mkdir(0o700, parents=True, exist_ok=True)
 else:
 self.local_shared_path = None
 defer.ensureDeferred(self._post_init())
 @property
 def version(self):
 self.bridge.register_method("profile_name_get", self.memory.get_profile_name)
 self.bridge.register_method("profiles_list_get", self.memory.get_profiles_list)
 self.bridge.register_method("entity_data_get", self.memory._get_entity_data)
 self.bridge.register_method("entities_data_get", self.memory._get_entities_data)
 self.bridge.register_method("profile_create", self.memory.create_profile)
-self.bridge.register_method("profile_delete_async", self.memory.profile_delete_async)
+self.bridge.register_method(
+"profile_delete_async", self.memory.profile_delete_async
+)
 self.bridge.register_method("profile_start_session", self.memory.start_session)
 self.bridge.register_method(
 "profile_is_session_started", self.memory._is_session_started
 )
-self.bridge.register_method("profile_set_default", self.memory.profile_set_default)
+self.bridge.register_method(
+"profile_set_default", self.memory.profile_set_default
+)
 self.bridge.register_method("connect", self._connect)
 self.bridge.register_method("disconnect", self.disconnect)
 self.bridge.register_method("contact_get", self._contact_get)
 self.bridge.register_method("contacts_get", self.contacts_get)
-self.bridge.register_method("contacts_get_from_group", self.contacts_get_from_group)
+self.bridge.register_method(
+"contacts_get_from_group", self.contacts_get_from_group
+)
 self.bridge.register_method("main_resource_get", self.memory._get_main_resource)
 self.bridge.register_method(
 "presence_statuses_get", self.memory._get_presence_statuses
 )
 self.bridge.register_method("sub_waiting_get", self.memory.sub_waiting_get)
 self.bridge.register_method("message_send", self._message_send)
-self.bridge.register_method("message_encryption_start",
+self.bridge.register_method(
-self._message_encryption_start)
+"message_encryption_start", self._message_encryption_start
-self.bridge.register_method("message_encryption_stop",
+)
-self._message_encryption_stop)
+self.bridge.register_method(
-self.bridge.register_method("message_encryption_get",
+"message_encryption_stop", self._message_encryption_stop
-self._message_encryption_get)
+)
-self.bridge.register_method("encryption_namespace_get",
+self.bridge.register_method(
-self._encryption_namespace_get)
+"message_encryption_get", self._message_encryption_get
-self.bridge.register_method("encryption_plugins_get", self._encryption_plugins_get)
+)
-self.bridge.register_method("encryption_trust_ui_get", self._encryption_trust_ui_get)
+self.bridge.register_method(
+"encryption_namespace_get", self._encryption_namespace_get
+)
+self.bridge.register_method(
+"encryption_plugins_get", self._encryption_plugins_get
+)
+self.bridge.register_method(
+"encryption_trust_ui_get", self._encryption_trust_ui_get
+)
 self.bridge.register_method("config_get", self._get_config)
 self.bridge.register_method("param_set", self.param_set)
 self.bridge.register_method("param_get_a", self.memory.get_string_param_a)
 self.bridge.register_method("private_data_get", self.memory._private_data_get)
 self.bridge.register_method("private_data_set", self.memory._private_data_set)
-self.bridge.register_method("private_data_delete", self.memory._private_data_delete)
+self.bridge.register_method(
-self.bridge.register_method("param_get_a_async", self.memory.async_get_string_param_a)
+"private_data_delete", self.memory._private_data_delete
+)
+self.bridge.register_method(
+"param_get_a_async", self.memory.async_get_string_param_a
+)
 self.bridge.register_method(
 "params_values_from_category_get_async",
 self.memory._get_params_values_from_category,
 )
 self.bridge.register_method("param_ui_get", self.memory._get_params_ui)
 self.bridge.register_method(
 "params_categories_get", self.memory.params_categories_get
 )
-self.bridge.register_method("params_register_app", self.memory.params_register_app)
+self.bridge.register_method(
+"params_register_app", self.memory.params_register_app
+)
 self.bridge.register_method("history_get", self.memory._history_get)
 self.bridge.register_method("presence_set", self._set_presence)
 self.bridge.register_method("subscription", self.subscription)
 self.bridge.register_method("contact_add", self._add_contact)
 self.bridge.register_method("contact_update", self._update_contact)
 self.bridge.register_method("session_infos_get", self.get_session_infos)
 self.bridge.register_method("devices_infos_get", self._get_devices_infos)
 self.bridge.register_method("namespaces_get", self.get_namespaces)
 self.bridge.register_method("image_check", self._image_check)
 self.bridge.register_method("image_resize", self._image_resize)
-self.bridge.register_method("image_generate_preview", self._image_generate_preview)
+self.bridge.register_method(
+"image_generate_preview", self._image_generate_preview
+)
 self.bridge.register_method("image_convert", self._image_convert)
 self.bridge.register_method("notification_add", self.memory._add_notification)
 self.bridge.register_method("notifications_get", self.memory._get_notifications)
-self.bridge.register_method("notification_delete", self.memory._delete_notification)
+self.bridge.register_method(
-self.bridge.register_method("notifications_expired_clean", self.memory._notifications_expired_clean)
+"notification_delete", self.memory._delete_notification
+)
+self.bridge.register_method(
+"notifications_expired_clean", self.memory._notifications_expired_clean
+)
 await self.memory.initialise()
 self.common_cache = cache.Cache(self, None)
 log.info(_("Memory initialised"))
 try:
 log.error(f"Init script is not a file: {init_script}")
 sys.exit(C.EXIT_BAD_ARG)
 else:
 log.info(f"Running init script {init_script!r}.")
 try:
-await async_process.run(
+await async_process.run(str(init_script), verbose=True)
-str(init_script),
-verbose=True
-)
 except RuntimeError as e:
 log.error(f"Init script failed: {e}")
 self.stopService()
 sys.exit(C.EXIT_ERROR)
 log.info(_("Backend is ready"))
 # profile autoconnection must be done after self.initialised is called because
 # start_session waits for it.
 autoconnect_dict = await self.memory.storage.get_ind_param_values(
-category='Connection', name='autoconnect_backend',
+category="Connection",
+name="autoconnect_backend",
 )
 profiles_autoconnect = [p for p, v in autoconnect_dict.items() if C.bool(v)]
 if not self.trigger.point("profilesAutoconnect", profiles_autoconnect):
 return
 if profiles_autoconnect:
-log.info(D_(
+log.info(
-"Following profiles will be connected automatically: {profiles}"
+D_(
-).format(profiles= ', '.join(profiles_autoconnect)))
+"Following profiles will be connected automatically: {profiles}"
+).format(profiles=", ".join(profiles_autoconnect))
+)
 connect_d_list = []
 for profile in profiles_autoconnect:
 connect_d_list.append(defer.ensureDeferred(self.connect(profile)))
 if connect_d_list:
 for idx, (success, result) in enumerate(results):
 if not success:
 profile = profiles_autoconnect[0]
 log.warning(
 _("Can't autoconnect profile {profile}: {reason}").format(
-profile = profile,
+profile=profile, reason=result
-reason = result)
+)
 )
 def _add_base_menus(self):
 """Add base menus"""
 encryption.EncryptionHandler._import_menus(self)
 for plug_path in plugins_path.glob("plugin_*"):
 if plug_path.is_dir():
 init_path = plug_path / f"__init__.{C.PLUGIN_EXT}"
 if not init_path.exists():
 log.warning(
-f"{plug_path} doesn't appear to be a package, can't load it")
+f"{plug_path} doesn't appear to be a package, can't load it"
+)
 continue
 plug_name = plug_path.name
 elif plug_path.is_file():
 if plug_path.suffix != f".{C.PLUGIN_EXT}":
 continue
 plug_name = plug_path.stem
 else:
-log.warning(
+log.warning(f"{plug_path} is not a file or a dir, ignoring it")
-f"{plug_path} is not a file or a dir, ignoring it")
 continue
 if not plug_name.isidentifier():
 log.warning(
-f"{plug_name!r} is not a valid name for a plugin, ignoring it")
+f"{plug_name!r} is not a valid name for a plugin, ignoring it"
+)
 continue
 plugin_path = f"libervia.backend.plugins.{plug_name}"
 try:
 __import__(plugin_path)
 except exceptions.MissingModule as e:
 self._unimport_plugin(plugin_path)
 log.warning(
 "Can't import plugin [{path}] because of an unavailale third party "
-"module:\n{msg}".format(
+"module:\n{msg}".format(path=plugin_path, msg=e)
-path=plugin_path, msg=e
-)
 )
 continue
 except exceptions.CancelError as e:
 log.info(
 "Plugin [{path}] cancelled its own import: {msg}".format(
 if not import_name:
 import_name, (plugin_path, mod, plugin_info) = plugins_to_import.popitem()
 else:
 if not import_name in plugins_to_import:
 if optional:
-log.warning(
+log.warning(_("Recommended plugin not found: {}").format(import_name))
-_("Recommended plugin not found: {}").format(import_name)
-)
 return
 msg = "Dependency not found: {}".format(import_name)
 log.error(msg)
 raise ImportError(msg)
 plugin_path, mod, plugin_info = plugins_to_import.pop(import_name)
 def _connect(self, profile_key, password="", options=None):
 profile = self.memory.get_profile_name(profile_key)
 return defer.ensureDeferred(self.connect(profile, password, options))
 async def connect(
-self, profile, password="", options=None, max_retries=C.XMPP_MAX_RETRIES):
+self, profile, password="", options=None, max_retries=C.XMPP_MAX_RETRIES
+):
 """Connect a profile (i.e. connect client.component to XMPP server)
 Retrieve the individual parameters, authenticate the profile
 and initiate the connection to the associated XMPP server.
 @param profile: %(doc_profile)s
 def get_namespace(self, short_name):
 try:
 return self.ns_map[short_name]
 except KeyError:
-raise exceptions.NotFound("namespace {short_name} is not registered"
+raise exceptions.NotFound(
-.format(short_name=short_name))
+"namespace {short_name} is not registered".format(short_name=short_name)
+)
 def get_session_infos(self, profile_key):
 """compile interesting data on current profile session"""
 client = self.get_client(profile_key)
-data = {
+data = {"jid": client.jid.full(), "started": str(int(client.started))}
-"jid": client.jid.full(),
-"started": str(int(client.started))
-}
 return defer.succeed(data)
 def _get_devices_infos(self, bare_jid, profile_key):
 client = self.get_client(profile_key)
 if not bare_jid:
 cache_data = self.memory.entity_data_get(client, res_jid)
 res_data = {
 "resource": resource,
 }
 try:
-presence = cache_data['presence']
+presence = cache_data["presence"]
 except KeyError:
 pass
 else:
-res_data['presence'] = {
+res_data["presence"] = {
 "show": presence.show,
 "priority": presence.priority,
 "statuses": presence.statuses,
 }
 disco = await self.get_disco_infos(client, res_jid)
 for (category, type_), name in disco.identities.items():
-identities = res_data.setdefault('identities', [])
+identities = res_data.setdefault("identities", [])
-identities.append({
+identities.append(
-"name": name,
+{
-"category": category,
+"name": name,
-"type": type_,
+"category": category,
-})
+"type": type_,
+}
+)
 ret_data.append(res_data)
 return ret_data
 @param path(Path): path to the image
 @return (Path): path to the generated preview
 """
 report = image.check(self, path, max_size=(300, 300))
-if not report['too_large']:
+if not report["too_large"]:
 # in the unlikely case that image is already smaller than a preview
 preview_path = path
 else:
 # we use hash as id, to re-use potentially existing preview
 path_hash = hashlib.sha256(str(path).encode()).hexdigest()
 uid = f"{path.stem}_{path_hash}_preview"
 filename = f"{uid}{path.suffix.lower()}"
 metadata = client.cache.get_metadata(uid=uid)
 if metadata is not None:
-preview_path = metadata['path']
+preview_path = metadata["path"]
 else:
 with client.cache.cache_data(
-source='HOST_PREVIEW',
+source="HOST_PREVIEW", uid=uid, filename=filename
-uid=uid,
+) as cache_f:
-filename=filename) as cache_f:
 preview_path = await image.resize(
-path,
+path, new_size=report["recommended_size"], dest=cache_f
-new_size=report['recommended_size'],
-dest=cache_f
 )
 return preview_path
 def _image_convert(self, source, dest, extra, profile_key):
 else:
 cache = client.cache
 metadata = cache.get_metadata(uid=uid)
 if metadata is not None:
 # there is already a conversion for this image in cache
-return metadata['path']
+return metadata["path"]
 else:
 with cache.cache_data(
-source='HOST_IMAGE_CONVERT',
+source="HOST_IMAGE_CONVERT", uid=uid, filename=filename
-uid=uid,
+) as cache_f:
-filename=filename) as cache_f:
 converted_path = await image.convert(
-source,
+source, dest=cache_f, extra=extra
-dest=cache_f,
-extra=extra
 )
 return converted_path
 else:
 return await image.convert(source, dest, extra)
 # local dirs
 def get_local_path(
 self,
 ## Encryption ##
 def register_encryption_plugin(self, *args, **kwargs):
 return encryption.EncryptionHandler.register_plugin(*args, **kwargs)
-def _message_encryption_start(self, to_jid_s, namespace, replace=False,
+def _message_encryption_start(
-profile_key=C.PROF_KEY_NONE):
+self, to_jid_s, namespace, replace=False, profile_key=C.PROF_KEY_NONE
+):
 client = self.get_client(profile_key)
 to_jid = jid.JID(to_jid_s)
 return defer.ensureDeferred(
-client.encryption.start(to_jid, namespace or None, replace))
+client.encryption.start(to_jid, namespace or None, replace)
+)
 def _message_encryption_stop(self, to_jid_s, profile_key=C.PROF_KEY_NONE):
 client = self.get_client(profile_key)
 to_jid = jid.JID(to_jid_s)
-return defer.ensureDeferred(
+return defer.ensureDeferred(client.encryption.stop(to_jid))
-client.encryption.stop(to_jid))
 def _message_encryption_get(self, to_jid_s, profile_key=C.PROF_KEY_NONE):
 client = self.get_client(profile_key)
 to_jid = jid.JID(to_jid_s)
 session_data = client.encryption.getSession(to_jid)
 def _encryption_plugins_get(self):
 plugins = encryption.EncryptionHandler.getPlugins()
 ret = []
 for p in plugins:
-ret.append({
+ret.append(
-"name": p.name,
+{
-"namespace": p.namespace,
+"name": p.name,
-"priority": p.priority,
+"namespace": p.namespace,
-"directed": p.directed,
+"priority": p.priority,
-})
+"directed": p.directed,
+}
+)
 return data_format.serialise(ret)
 def _encryption_trust_ui_get(self, to_jid_s, namespace, profile_key):
 client = self.get_client(profile_key)
 to_jid = jid.JID(to_jid_s)
 d = defer.ensureDeferred(
-client.encryption.get_trust_ui(to_jid, namespace=namespace or None))
+client.encryption.get_trust_ui(to_jid, namespace=namespace or None)
+)
 d.addCallback(lambda xmlui: xmlui.toXml())
 return d
 ## XMPP methods ##
 def _message_send(
-self, to_jid_s, message, subject=None, mess_type="auto", extra_s="",
+self,
-profile_key=C.PROF_KEY_NONE):
+to_jid_s,
+message,
+subject=None,
+mess_type="auto",
+extra_s="",
+profile_key=C.PROF_KEY_NONE,
+):
 client = self.get_client(profile_key)
 to_jid = jid.JID(to_jid_s)
 return client.sendMessage(
-to_jid,
+to_jid, message, subject, mess_type, data_format.deserialise(extra_s)
-message,
-subject,
-mess_type,
-data_format.deserialise(extra_s)
 )
 def _set_presence(self, to="", show="", statuses=None, profile_key=C.PROF_KEY_NONE):
 return self.presence_set(jid.JID(to) if to else None, show, statuses, profile_key)
-def presence_set(self, to_jid=None, show="", statuses=None,
+def presence_set(
-profile_key=C.PROF_KEY_NONE):
+self, to_jid=None, show="", statuses=None, profile_key=C.PROF_KEY_NONE
+):
 """Send our presence information"""
 if statuses is None:
 statuses = {}
 profile = self.memory.get_profile_name(profile_key)
 assert profile
 return self.contact_update(client, jid.JID(to_jid_s), name, groups)
 def contact_update(self, client, to_jid, name, groups):
 """update a contact in roster list"""
 roster_item = RosterItem(to_jid)
-roster_item.name = name or u''
+roster_item.name = name or ""
 roster_item.groups = set(groups)
 if not self.trigger.point("roster_update", client, roster_item):
 return
 return client.roster.setItem(roster_item)
 return self.memory.disco.find_service_entities(*args, **kwargs)
 def find_features_set(self, *args, **kwargs):
 return self.memory.disco.find_features_set(*args, **kwargs)
-def _find_by_features(self, namespaces, identities, bare_jids, service, roster, own_jid,
+def _find_by_features(
-local_device, profile_key):
+self,
+namespaces,
+identities,
+bare_jids,
+service,
+roster,
+own_jid,
+local_device,
+profile_key,
+):
 client = self.get_client(profile_key)
 identities = [tuple(i) for i in identities] if identities else None
-return defer.ensureDeferred(self.find_by_features(
+return defer.ensureDeferred(
-client, namespaces, identities, bare_jids, service, roster, own_jid,
+self.find_by_features(
-local_device))
+client,
+namespaces,
+identities,
+bare_jids,
+service,
+roster,
+own_jid,
+local_device,
+)
+)
 async def find_by_features(
 self,
 client: SatXMPPEntity,
 namespaces: List[str],
-identities: Optional[List[Tuple[str, str]]]=None,
+identities: Optional[List[Tuple[str, str]]] = None,
-bare_jids: bool=False,
+bare_jids: bool = False,
-service: bool=True,
+service: bool = True,
-roster: bool=True,
+roster: bool = True,
-own_jid: bool=True,
+own_jid: bool = True,
-local_device: bool=False
+local_device: bool = False,
 ) -> Tuple[
 Dict[jid.JID, Tuple[str, str, str]],
 Dict[jid.JID, Tuple[str, str, str]],
-Dict[jid.JID, Tuple[str, str, str]]
+Dict[jid.JID, Tuple[str, str, str]],
 ]:
 """Retrieve all services or contacts managing a set a features
 @param namespaces: features which must be handled
 @param identities: if not None or empty,
 found_own = {}
 found_roster = {}
 if service:
 services_jids = await self.find_features_set(client, namespaces)
 services_jids = list(services_jids)  # we need a list to map results below
-services_infos  = await defer.DeferredList(
+services_infos = await defer.DeferredList(
-[self.get_disco_infos(client, service_jid) for service_jid in services_jids]
+[
+self.get_disco_infos(client, service_jid)
+for service_jid in services_jids
+]
 )
 for idx, (success, infos) in enumerate(services_infos):
 service_jid = services_jids[idx]
 if not success:
 log.warning(
-_("Can't find features for service {service_jid}, ignoring")
+_(
-.format(service_jid=service_jid.full()))
+"Can't find features for service {service_jid}, ignoring"
+).format(service_jid=service_jid.full())
+)
 continue
-if (identities is not None
+if identities is not None and not set(infos.identities.keys()).issuperset(
-and not set(infos.identities.keys()).issuperset(identities)):
+identities
+):
 continue
 found_identities = [
 (cat, type_, name or "")
 for (cat, type_), name in infos.identities.items()
 ]
 for idx, (success, infos) in enumerate(infos_data):
 full_jid = full_jids[idx]
 if not success:
 log.warning(
-_("Can't retrieve {full_jid} infos, ignoring")
+_("Can't retrieve {full_jid} infos, ignoring").format(
-.format(full_jid=full_jid.full()))
+full_jid=full_jid.full()
+)
+)
 continue
 if infos.features.issuperset(namespaces):
 if identities is not None and not set(
 infos.identities.keys()
 ).issuperset(identities):
 reactor.callLater(1800, purge_callback)
 return callback_id
 def remove_callback(self, callback_id):
-""" Remove a previously registered callback
+"""Remove a previously registered callback
-@param callback_id: id returned by [register_callback] """
+@param callback_id: id returned by [register_callback]"""
 log.debug("Removing callback [%s]" % callback_id)
 del self._cb_map[callback_id]
 def _action_launch(
-self,
+self, callback_id: str, data_s: str, profile_key: str
-callback_id: str,
-data_s: str,
-profile_key: str
 ) -> defer.Deferred:
 d = self.launch_callback(
-callback_id,
+callback_id, data_format.deserialise(data_s), profile_key
-data_format.deserialise(data_s),
-profile_key
 )
 d.addCallback(data_format.serialise)
 return d
 def launch_callback(
 self,
 callback_id: str,
 data: Optional[dict] = None,
-profile_key: str = C.PROF_KEY_NONE
+profile_key: str = C.PROF_KEY_NONE,
 ) -> defer.Deferred:
 """Launch a specific callback
 @param callback_id: id of the action (callback) to launch
 @param data: optional data
 @param path(iterable[unicode]): untranslated path to menu
 @return (tuple[unicode]): canonical form of path
 """
 return tuple((p.lower().strip() for p in path))
-def import_menu(self, path, callback, security_limit=C.NO_SECURITY_LIMIT,
+def import_menu(
-help_string="", type_=C.MENU_GLOBAL):
+self,
+path,
+callback,
+security_limit=C.NO_SECURITY_LIMIT,
+help_string="",
+type_=C.MENU_GLOBAL,
+):
 r"""register a new menu for frontends
 @param path(iterable[unicode]): path to go to the menu
 (category/subcategory/.../item) (e.g.: ("File", "Open"))
 /!\ use D_() instead of _() for translations (e.g. (D_("File"), D_("Open")))
 extra = {}  # TODO: manage extra data like icon
 ret.append((menu_id, type_, path, path_i18n, extra))
 return ret
-def _launch_menu(self, menu_type, path, data=None, security_limit=C.NO_SECURITY_LIMIT,
+def _launch_menu(
-profile_key=C.PROF_KEY_NONE):
+self,
+menu_type,
+path,
+data=None,
+security_limit=C.NO_SECURITY_LIMIT,
+profile_key=C.PROF_KEY_NONE,
+):
 client = self.get_client(profile_key)
 return self.launch_menu(client, menu_type, path, data, security_limit)
-def launch_menu(self, client, menu_type, path, data=None,
+def launch_menu(
-security_limit=C.NO_SECURITY_LIMIT):
+self, client, menu_type, path, data=None, security_limit=C.NO_SECURITY_LIMIT
+):
 """launch action a menu action
 @param menu_type(unicode): type of menu to launch
 @param path(iterable[unicode]): canonical path of the menu
 @params data(dict): menu data

Mercurial > libervia-backend

comparison libervia/backend/core/main.py @ 4270:0d7bb4df2343