libervia-backend: libervia/backend/plugins/plugin_misc_app_manager/__init_

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

Reformatted code base using black.

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

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 try:
 import yaml
 except ImportError:
 raise exceptions.MissingModule(
-'Missing module PyYAML, please download/install it. You can use '
+"Missing module PyYAML, please download/install it. You can use "
 '"pip install pyyaml"'
 )
 try:
 from yaml import CLoader as Loader, CDumper as Dumper
 log.warning(
 "Can't use LibYAML binding (is libyaml installed?), pure Python version will be "
 "used, but it is slower"
 )
 from yaml import Loader, Dumper
 PLUGIN_INFO = {
 C.PI_NAME: "Applications Manager",
 C.PI_IMPORT_NAME: "APP_MANAGER",
 C.PI_DESCRIPTION: _(
 """Applications Manager
 Manage external applications using packagers, OS virtualization/containers or other
 software management tools.
-"""),
+"""
+),
 }
 APP_FILE_PREFIX = "libervia_app_"
 self._apps = {}
 self._started = {}
 # instance id to app data map
 self._instances = {}
 self.persistent_data = persistent.LazyPersistentBinaryDict("app_manager")
 host.bridge.add_method(
 "applications_list",
 ".plugin",
 method=self._get_exposed,
 async_=True,
 )
 # application has been started succeesfully,
 # args: name, instance_id, extra
-host.bridge.add_signal(
+host.bridge.add_signal("application_started", ".plugin", signature="sss")
-"application_started", ".plugin", signature="sss"
-)
 # application went wrong with the application
 # args: name, instance_id, extra
-host.bridge.add_signal(
+host.bridge.add_signal("application_error", ".plugin", signature="sss")
-"application_error", ".plugin", signature="sss"
+yaml.add_constructor("!libervia_conf", self._libervia_conf_constr, Loader=Loader)
-)
 yaml.add_constructor(
-"!libervia_conf", self._libervia_conf_constr, Loader=Loader)
+"!libervia_generate_pwd", self._libervia_generate_pwd_constr, Loader=Loader
+)
 yaml.add_constructor(
-"!libervia_generate_pwd", self._libervia_generate_pwd_constr, Loader=Loader)
+"!libervia_param", self._libervia_param_constr, Loader=Loader
-yaml.add_constructor(
+)
-"!libervia_param", self._libervia_param_constr, Loader=Loader)
 def unload(self):
 log.debug("unloading applications manager")
 for instances in self._started.values():
 for instance in instances:
-data = instance['data']
+data = instance["data"]
-if not data['single_instance']:
+if not data["single_instance"]:
 log.debug(
-f"cleaning temporary directory at {data['_instance_dir_path']}")
+f"cleaning temporary directory at {data['_instance_dir_path']}"
-data['_instance_dir_obj'].cleanup()
+)
+data["_instance_dir_obj"].cleanup()
 def _libervia_conf_constr(self, loader, node) -> str:
 """Get a value from Libervia configuration
 A list is expected with either "name" of a config parameter, a one or more of
 "is expected"
 )
 value = self.host.memory.config_get(section, name, default)
 # FIXME: "public_url" is used only here and doesn't take multi-sites into account
-if name == "public_url" and (not value or value.startswith('http')):
+if name == "public_url" and (not value or value.startswith("http")):
 if not value:
-log.warning(_(
+log.warning(
-'No value found for "public_url", using "example.org" for '
+_(
-'now, please set the proper value in libervia.conf'))
+'No value found for "public_url", using "example.org" for '
+"now, please set the proper value in libervia.conf"
+)
+)
 else:
-log.warning(_(
+log.warning(
-'invalid value for "public_url" ({value}), it musts not start with '
+_(
-'schema ("http"), ignoring it and using "example.org" '
+'invalid value for "public_url" ({value}), it musts not start with '
-'instead')
+'schema ("http"), ignoring it and using "example.org" '
-.format(value=value))
+"instead"
+).format(value=value)
+)
 value = "example.org"
 if filter_ is None:
 pass
-elif filter_ == 'first':
+elif filter_ == "first":
 value = value[0]
 elif filter_ == "not":
 value = C.bool(value)
 value = C.bool_const(not value).lower()
 else:
 try:
 key = self._app_persistent_data[pwd_data_key]
 except KeyError:
 alphabet = string.ascii_letters + string.digits
 key_size = int(kwargs.get("size", 30))
-key = ''.join(secrets.choice(alphabet) for __ in range(key_size))
+key = "".join(secrets.choice(alphabet) for __ in range(key_size))
 self._app_persistent_data[pwd_data_key] = key
 else:
 log.debug(f"Re-using existing key for {name!r} password.")
 return key
 def register(self, manager):
 name = manager.name
 if name in self._managers:
 raise exceptions.ConflictError(
-f"There is already a manager with the name {name}")
+f"There is already a manager with the name {name}"
+)
 self._managers[manager.name] = manager
 if manager.discover_path is not None:
 self.discover(manager.discover_path, manager)
 def get_manager(self, app_data: dict) -> AppManagerBackend:
 @raise exceptions.NotFound: manager is not registered
 """
 try:
 app_type = app_data["type"]
 except KeyError:
-raise exceptions.DataError(
+raise exceptions.DataError('app file doesn\'t have the mandatory "type" key')
-"app file doesn't have the mandatory \"type\" key"
-)
 if not isinstance(app_type, str):
-raise exceptions.DataError(
+raise exceptions.DataError(f"invalid app data type: {app_type!r}")
-f"invalid app data type: {app_type!r}"
-)
 app_type = app_type.strip()
 try:
 return self._managers[app_type]
 except KeyError:
 raise exceptions.NotFound(
-f"No manager found to manage app of type {app_type!r}")
+f"No manager found to manage app of type {app_type!r}"
+)
-def get_app_data(
-self,
+def get_app_data(self, id_type: Optional[str], identifier: str) -> dict:
-id_type: Optional[str],
-identifier: str
-) -> dict:
 """Retrieve instance's app_data from identifier
 @param id_type: type of the identifier, can be:
 - "name": identifier is a canonical application name
 the first found instance of this application is returned
 @return: instance application data
 @raise exceptions.NotFound: no instance with this id can be found
 @raise ValueError: id_type is invalid
 """
 if not id_type:
-id_type = 'name'
+id_type = "name"
-if id_type == 'name':
+if id_type == "name":
 identifier = identifier.lower().strip()
 try:
 return next(iter(self._started[identifier]))
 except (KeyError, StopIteration):
 raise exceptions.NotFound(
 f"No instance of {identifier!r} is currently running"
 )
-elif id_type == 'instance':
+elif id_type == "instance":
 instance_id = identifier
 try:
 return self._instances[instance_id]
 except KeyError:
 raise exceptions.NotFound(
 f"There is no application instance running with id {instance_id!r}"
 )
 else:
 raise ValueError(f"invalid id_type: {id_type!r}")
-def discover(
+def discover(self, dir_path: Path, manager: AppManagerBackend | None = None) -> None:
-self,
-dir_path: Path,
-manager: AppManagerBackend|None = None
-) -> None:
 """Search for app configuration file.
 App configuration files must start with [APP_FILE_PREFIX] and have a ``.yaml``
 extension.
 """
 if manager is None:
 try:
 app_data = self.parse(file_path)
 manager = self.get_manager(app_data)
 except (exceptions.DataError, exceptions.NotFound) as e:
-log.warning(
+log.warning(f"Can't parse {file_path}, skipping: {e}")
-f"Can't parse {file_path}, skipping: {e}")
 continue
-app_name = file_path.stem[len(APP_FILE_PREFIX):].strip().lower()
+app_name = file_path.stem[len(APP_FILE_PREFIX) :].strip().lower()
 if not app_name:
 log.warning(f"invalid app file name at {file_path}")
 continue
 app_dict = self._apps.setdefault(app_name, {})
 manager_set = app_dict.setdefault(manager, set())
 manager_set.add(file_path)
-log.debug(
+log.debug(f"{app_name!r} {manager.name} application found")
-f"{app_name!r} {manager.name} application found"
-)
 def parse(self, file_path: Path, params: Optional[dict] = None) -> dict:
 """Parse Libervia application file
 @param params: parameters for running this instance
 app_data = self.load(f)
 self._params = None
 if "name" not in app_data:
 # note that we don't use lower() here as we want human readable name and
 # uppercase may be set on purpose
-app_data['name'] = file_path.stem[len(APP_FILE_PREFIX):].strip()
+app_data["name"] = file_path.stem[len(APP_FILE_PREFIX) :].strip()
 single_instance = app_data.setdefault("single_instance", True)
 if not isinstance(single_instance, bool):
 raise ValueError(
 f'"single_instance" must be a boolean, but it is {type(single_instance)}'
 )
 #   to be defined in app_data allows explicitly it.
 app_name = app_name.lower().strip()
 try:
 app_file_path = next(iter(next(iter(self._apps[app_name].values()))))
 except KeyError:
-raise exceptions.NotFound(
+raise exceptions.NotFound(f"No application found with the name {app_name!r}")
-f"No application found with the name {app_name!r}"
-)
 log.info(f"starting {app_name!r}")
 self._app_persistent_data = await self.persistent_data.get(app_name) or {}
 self._app_persistent_data["last_started"] = time.time()
 started_data = self._started.setdefault(app_name, [])
 app_data = self.parse(app_file_path, extra)
 await self.persistent_data.aset(app_name, self._app_persistent_data)
 app_data["_started"] = False
-app_data['_file_path'] = app_file_path
+app_data["_file_path"] = app_file_path
-app_data['_name_canonical'] = app_name
+app_data["_name_canonical"] = app_name
-single_instance = app_data['single_instance']
+single_instance = app_data["single_instance"]
-ret_data = {
+ret_data = {"name": app_name, "started": False}
-"name": app_name,
-"started": False
-}
 if single_instance:
 if started_data:
 instance_data = started_data[0]
 instance_id = instance_data["_instance_id"]
 ret_data["instance"] = instance_id
 else:
 cache_path = self.host.memory.get_cache_path(
 PLUGIN_INFO[C.PI_IMPORT_NAME], app_name
 )
 cache_path.mkdir(0o700, parents=True, exist_ok=True)
-app_data['_instance_dir_path'] = cache_path
+app_data["_instance_dir_path"] = cache_path
 else:
 dest_dir_obj = tempfile.TemporaryDirectory(prefix="libervia_app_")
-app_data['_instance_dir_obj'] = dest_dir_obj
+app_data["_instance_dir_obj"] = dest_dir_obj
-app_data['_instance_dir_path'] = Path(dest_dir_obj.name)
+app_data["_instance_dir_path"] = Path(dest_dir_obj.name)
-instance_id = ret_data["instance"] = app_data['_instance_id'] = shortuuid.uuid()
+instance_id = ret_data["instance"] = app_data["_instance_id"] = shortuuid.uuid()
 manager = self.get_manager(app_data)
-app_data['_manager'] = manager
+app_data["_manager"] = manager
 started_data.append(app_data)
 self._instances[instance_id] = app_data
 # we retrieve exposed data such as url_prefix which can be useful computed exposed
 # data must wait for the app to be started, so we skip them for now
 ret_data["expose"] = await self.get_exposed(
 try:
 start = manager.start
 except AttributeError:
 raise exceptions.InternalError(
-f"{manager.name} doesn't have the mandatory \"start\" method"
+f'{manager.name} doesn\'t have the mandatory "start" method'
 )
 else:
 defer.ensureDeferred(self.start_app(start, app_data))
 return ret_data
 except Exception as e:
 log.exception(f"Can't start libervia app {app_name!r}")
 self.host.bridge.application_error(
 app_name,
 instance_id,
-data_format.serialise(
+data_format.serialise({"class": str(type(e)), "msg": str(e)}),
-{
+)
-"class": str(type(e)),
-"msg": str(e)
-}
-))
 else:
 app_data["_started"] = True
 self.host.bridge.application_started(app_name, instance_id, "")
 log.info(f"{app_name!r} started")
 def _stop(self, identifier, id_type, extra):
 extra = data_format.deserialise(extra)
 return defer.ensureDeferred(
-self.stop(str(identifier), str(id_type) or None, extra))
+self.stop(str(identifier), str(id_type) or None, extra)
+)
 async def stop(
 self,
 identifier: str,
 id_type: Optional[str] = None,
 app_data = self.get_app_data(id_type, identifier)
 log.info(f"stopping {app_data['name']!r}")
-app_name = app_data['_name_canonical']
+app_name = app_data["_name_canonical"]
-instance_id = app_data['_instance_id']
+instance_id = app_data["_instance_id"]
-manager = app_data['_manager']
+manager = app_data["_manager"]
 try:
 stop = manager.stop
 except AttributeError:
 raise exceptions.InternalError(
-f"{manager.name} doesn't have the mandatory \"stop\" method"
+f'{manager.name} doesn\'t have the mandatory "stop" method'
 )
 else:
 try:
 await stop(app_data)
 except Exception as e:
 try:
 del self._instances[instance_id]
 except KeyError:
 log.error(
-f"INTERNAL ERROR: {instance_id!r} is not present in self._instances")
+f"INTERNAL ERROR: {instance_id!r} is not present in self._instances"
+)
 try:
 self._started[app_name].remove(app_data)
 except ValueError:
 log.error(
 The manager's "compute_expose" method will be called if it exists. It can be used
 to handle manager specific conventions.
 """
 app_data = self.get_app_data(id_type, identifier)
-if app_data.get('_exposed_computed', False):
+if app_data.get("_exposed_computed", False):
-return app_data['expose']
+return app_data["expose"]
 if extra is None:
 extra = {}
 expose = app_data.setdefault("expose", {})
 if "passwords" in expose:
-passwords = expose['passwords']
+passwords = expose["passwords"]
 for name, value in list(passwords.items()):
 if isinstance(value, list):
 # if we have a list, is the sequence of keys leading to the value
 # to expose.
 try:
 passwords[name] = self.get_app_data_value(value, app_data)
 except KeyError:
 log.warning(
-f"Can't retrieve exposed value for password {name!r}: {e}")
+f"Can't retrieve exposed value for password {name!r}: {e}"
+)
 del passwords[name]
 for key in ("url_prefix", "front_url"):
 value = expose.get(key)
 if isinstance(value, list):
 try:
 expose[key] = self.get_app_data_value(value, app_data)
 except KeyError:
-log.warning(
+log.warning(f"Can't retrieve exposed value for {key!r} at {value}")
-f"Can't retrieve exposed value for {key!r} at {value}"
-)
 del expose[key]
 front_url = expose.get("front_url")
 if front_url:
 # we want to be sure that a scheme is defined, defaulting to ``https``
 if not parsed_url.scheme:
 if not parsed_url.netloc:
 path_elt = parsed_url.path.split("/", 1)
 parsed_url = parsed_url._replace(
 netloc=path_elt[0],
-path=f"/{path_elt[1]}" if len(path_elt) > 1 else ""
+path=f"/{path_elt[1]}" if len(path_elt) > 1 else "",
 )
-parsed_url = parsed_url._replace(scheme='https')
+parsed_url = parsed_url._replace(scheme="https")
 expose["front_url"] = urlunparse(parsed_url)
 if extra.get("skip_compute", False):
 return expose
 try:
-compute_expose = app_data['_manager'].compute_expose
+compute_expose = app_data["_manager"].compute_expose
 except AttributeError:
 pass
 else:
 await compute_expose(app_data)
-app_data['_exposed_computed'] = True
+app_data["_exposed_computed"] = True
 return expose
 async def _do_prepare(
 self,
 app_data: dict,
 ) -> None:
-name = app_data['name']
+name = app_data["name"]
-dest_path = app_data['_instance_dir_path']
+dest_path = app_data["_instance_dir_path"]
 if next(dest_path.iterdir(), None) != None:
 log.debug(f"There is already a prepared dir at {dest_path}, nothing to do")
 return
 try:
-prepare = app_data['prepare'].copy()
+prepare = app_data["prepare"].copy()
 except KeyError:
 prepare = {}
 if not prepare:
 log.debug(f"Nothing to prepare for {name!r}")
 for action, value in list(prepare.items()):
 log.debug(f"[{name}] [prepare] running {action!r} action")
 if action == "git":
 try:
-git_path = which('git')[0]
+git_path = which("git")[0]
 except IndexError:
 raise exceptions.NotFound(
 "Can't find \"git\" executable, {name} can't be started without it"
 )
 await async_process.run(git_path, "clone", value, str(dest_path))
 async def _do_create_files(
 self,
 app_data: dict,
 ) -> None:
-dest_path = app_data['_instance_dir_path']
+dest_path = app_data["_instance_dir_path"]
-files = app_data.get('files')
+files = app_data.get("files")
 if not files:
 return
 if not isinstance(files, dict):
 raise ValueError('"files" must be a dictionary')
 for filename, data in files.items():

Mercurial > libervia-backend

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

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