# Copyright (c) 2016 Weitian LI # MIT license """ Handle the configurations operations with the client. """ import os import logging from ...errors import ConfigError logger = logging.getLogger(__name__) class ConfigsHandler: """ Handle the "configs" type of messages from the client. """ def __init__(self, configs): self.configs = configs def handle_message(self, msg): """ Handle the message of type "configs", which request to get or set some configurations by the client. TODO: * improve the description ... * split these handling functions into a separate class in a module Parameters ---------- msg : dict A dictionary parsed from the incoming JSON message, which generally has the following syntax: ``{"type": "configs", "action": , "data": }`` where the ```` is ``set`` or ``get``, and the ```` is a list of config keys or a dict of config key-value pairs. Returns ------- response : dict A dictionary parsed from the incoming JSON message, which generally has the following syntax: ``{"type": "configs", "action": , "data": , "errors": }`` where the ```` is the same as input, the ```` is a list of config keys or a dict of config key-value pairs, and ```` contains the error message for the invalid config values. """ try: msg_type = msg["type"] msg_action = msg["action"] response = {"type": msg_type, "action": msg_action} logger.info("WebSocket: handle message: " + "type: {0}, action: {1}".format(msg_type, msg_action)) if msg_action == "get": # Get the values of the specified options try: data, errors = self._get_configs(keys=msg["keys"]) response["success"] = True response["data"] = data response["errors"] = errors except KeyError: response["success"] = False response["error"] = "'keys' is missing" elif msg_action == "set": # Set the values of the specified options try: errors = self._set_configs(data=msg["data"]) response["success"] = True response["data"] = {} # be more consistent response["errors"] = errors except KeyError: response["success"] = False response["error"] = "'data' is missing" elif msg_action == "reset": # Reset the configurations to the defaults self._reset_configs() response["success"] = True elif msg_action == "load": # Load the supplied user configuration file try: success, error = self._load_configs(msg["userconfig"]) response["success"] = success if not success: response["error"] = error except KeyError: response["success"] = False response["error"] = "'userconfig' is missing" elif msg_action == "save": # Save current configurations to file try: success, error = self._save_configs(msg["outfile"], msg["clobber"]) response["success"] = success if not success: response["error"] = error except KeyError: response["success"] = False response["error"] = "'outfile' or 'clobber' is missing" else: logger.warning("WebSocket: " + "unknown action: {0}".format(msg_action)) response["success"] = False response["error"] = "unknown action: {0}".format(msg_action) except KeyError: # Received message has wrong syntax/format response = {"success": False, "type": msg_type, "error": "no action specified"} # logger.debug("WebSocket: response: {0}".format(response)) return response def _get_configs(self, keys=None): """Get the values of the config options specified by the given keys. Parameters ---------- keys : list[str], optional A list of keys specifying the config options whose values will be obtained. If ``keys=None``, then all the configurations values are dumped. Returns ------- data : dict A dictionary with keys the same as the input keys, and values the corresponding config option values. errors : dict When error occurs (e.g., invalid key), then the specific errors with details are stored in this dictionary. NOTE ---- Do not forget the ``userconfig`` option. """ if keys is None: # Dump all the configurations data = self.configs.dump(flatten=True) data["userconfig"] = self.configs.userconfig errors = {} else: data = {} errors = {} for key in keys: if key == "userconfig": data["userconfig"] = self.configs.userconfig else: try: data[key] = self.configs.getn(key) except KeyError as e: errors[key] = str(e) # return (data, errors) def _set_configs(self, data): """Set the values of the config options specified by the given keys to the corresponding supplied data. NOTE ---- The ``userconfig`` needs special handle. The ``workdir`` and ``configfile`` options should be ignored. Parameters ---------- data : dict A dictionary of key-value pairs, with keys specifying the config options whose value will be changed, and values the new values to which config options will be set. NOTE: If want to set the ``userconfig`` option, an *absolute path* must be provided. Returns ------- errors : dict When error occurs (e.g., invalid key, invalid values), then the specific errors with details are stored in this dictionary. """ errors = {} for key, value in data.items(): if key in ["workdir", "configfile"]: # Ignore "workdir" and "configfile" continue elif key == "userconfig": if os.path.isabs(os.path.expanduser(value)): self.configs.userconfig = value else: errors[key] = "Not an absolute path" else: try: self.configs.setn(key, value) except KeyError as e: errors[key] = str(e) # NOTE: # Check the whole configurations after all provided options are # updated, and merge the validation errors. __, cherr = self.configs.check_all(raise_exception=False) errors.update(cherr) return errors def _reset_configs(self): """Reset the configurations to the defaults.""" self.configs.reset() def _load_configs(self, userconfig): """Load configurations from the provided user configuration file. Parameters ---------- userconfig: str The filepath to the user configuration file, which must be an *absolute path*. Returns ------- success : bool ``True`` if the operation succeeded, otherwise, ``False``. error : str If failed, this ``error`` saves the details, otherwise, ``None``. """ success = False error = None if os.path.isabs(os.path.expanduser(userconfig)): try: self.configs.read_userconfig(userconfig) success = True except ConfigError as e: error = str(e) else: error = "Not an absolute path" return (success, error) def _save_configs(self, outfile, clobber=False): """Save current configurations to file. Parameters ---------- outfile: str The filepath to the output configuration file, which must be an *absolute path*. clobber : bool, optional Whether overwrite the output file if already exists? Returns ------- success : bool ``True`` if the operation succeeded, otherwise, ``False``. error : str If failed, this ``error`` saves the details, otherwise, ``None``. """ success = False error = None try: self.configs.save(outfile, clobber=clobber) success = True except (ValueError, OSError) as e: error = str(e) return (success, error)