Source code for pkglts.config_management

"""
Main config object and functions to manipulate it.
"""

import json
import logging
from datetime import date

from jinja2 import Environment, StrictUndefined, UndefinedError

from .config import pkg_cfg_file, pkglts_dir
from .option.git.option import OptionGit
from .option.pypi.option import OptionPypi
from .option.reqs.option import OptionReqs
from .option.src.option import OptionSrc
from .option_tools import available_options, find_available_options

CURRENT_PKG_CFG_VERSION = 18

LOGGER = logging.getLogger(__name__)

DEFAULT_CFG = dict(
    _pkglts=dict(use_prompts=False, auto_install=True, install_front_end="stdout", version=CURRENT_PKG_CFG_VERSION)
)

find_available_options()


[docs] class ConfigSection(object): """Small class to allow accessing parameters using the dot method instead of ['param_name'] method """ pass
[docs] class Config(dict): """Object used to store both a templated version of the config as a dict interface its resolution and a jinja2 environment that reflect the config. """ def __init__(self, *args, **kwds): dict.__init__(self) self._tpl = dict(*args, **kwds) # initialise associated Jinja2 environment self._env = Environment(undefined=StrictUndefined) self._env.keep_trailing_newline = True # add global filters and test self._env.globals["today"] = lambda: date.today().isoformat() self.add_test("available", self._is_available) # resolve self.resolve() def __eq__(self, other): return dict.__eq__(self, other) def __ne__(self, other): return dict.__ne__(self, other)
[docs] def template(self): """Associated template even after resolution.""" return self._tpl
[docs] def add_test(self, name, func): """Add a new test in jinja2 environment. Args: name (str): name of test (must be unique) func (callable): function use for test Returns: None """ self._env.tests[name] = func
def _is_available(self, opt_name): return opt_name in self def _add_param(self, opt_name, param_name, param_value): """Add a new parameter value in the config Args: opt_name (str): Option name param_name (str): parameter name param_value (any): parameter value Returns: None """ self[opt_name][param_name] = param_value setattr(self._env.globals[opt_name], param_name, param_value)
[docs] def resolve(self): """Try to resolve all templated items. Returns: None """ to_eval = [] for opt_name, cfg in self._tpl.items(): self[opt_name] = {} self._env.globals[opt_name] = ConfigSection() for key, param in cfg.items(): if isinstance(param, str): to_eval.append((opt_name, key, param)) else: self._add_param(opt_name, key, param) nb_iter_max = len(to_eval) ** 2 cur_iter = 0 while to_eval and cur_iter < nb_iter_max: cur_iter += 1 opt_name, key, param = to_eval.pop(0) try: txt = self.render(param) self._add_param(opt_name, key, txt) except UndefinedError: to_eval.append((opt_name, key, param)) if to_eval: msg = "unable to fully render config\n" for item in to_eval: msg += "{}:{} '{}'\n".format(*item) raise UserWarning(msg)
[docs] def load_extra(self): """load option specific handlers. Returns: None """ for opt_name in self.installed_options(): try: opt = available_options[opt_name] for func_name, func in opt.environment_extensions(self).items(): setattr(self._env.globals[opt_name], func_name, func) except KeyError: raise KeyError(f"option '{opt_name}' exists in config but does not appear to be installed")
[docs] def installed_options(self, return_sorted=False): """List all installed options. Returns: (iter of str) """ installed = [key for key in self if not key.startswith("_")] if return_sorted: opt_names = installed installed = [] while opt_names: name = opt_names.pop(0) opt = available_options[name] if any(dep not in installed for dep in opt.require_option(self)): opt_names.append(name) else: installed.append(name) for name in installed: yield name
[docs] def render(self, txt): """Use items in config to render text Args: txt (str): templated text to render Returns: (str): same text where all templated parts have been replaced by their values. """ return self._env.from_string(txt).render()
[docs] def get_pkg_config(rep="."): """Read pkg_cfg file associated to this package. Args: rep (Path): directory to search for info Returns: (Config): Config initialized with pkg_config """ with open(rep / pkglts_dir / pkg_cfg_file, "r") as fhr: pkg_cfg = json.load(fhr) # update version of pkg_config file_version = pkg_cfg["_pkglts"].get("version", 0) for i in range(file_version, CURRENT_PKG_CFG_VERSION): upgrade_pkg_cfg_version(pkg_cfg, i) # create Config object cfg = Config(pkg_cfg) cfg.load_extra() # write back config if version has been updated if file_version < CURRENT_PKG_CFG_VERSION: write_pkg_config(cfg, rep) return cfg
[docs] def write_pkg_config(cfg, rep="."): """Store config associated to this package on disk. Args: cfg (Config): current working config rep (Path): directory to search for info Returns: None """ LOGGER.info("write package config") pkg_cfg = dict(cfg.template()) with open(rep / pkglts_dir / pkg_cfg_file, "w") as fhw: json.dump(pkg_cfg, fhw, sort_keys=True, indent=2)
[docs] def upgrade_pkg_cfg_version(pkg_cfg, version): """Upgrade the version of pkg_cfg file from version to version +1 Args: pkg_cfg (dict of str, any): package configuration version (int): current version of file Returns: (dict of str: any): a reference to an updated pkg_cfg """ if version == 0: pkg_cfg["_pkglts"]["version"] = 1 elif version == 1: pkg_cfg["_pkglts"]["version"] = 2 elif version == 2: pkg_cfg["_pkglts"]["version"] = 3 if "pysetup" in pkg_cfg: section = pkg_cfg["pysetup"] section["require"] = section.get("require", []) elif version == 3: pkg_cfg["_pkglts"]["version"] = 4 if "test" in pkg_cfg: section = pkg_cfg["test"] section["suite_name"] = section.get("suite_name", "pytest") elif version == 4: pkg_cfg["_pkglts"]["version"] = 5 if "base" in pkg_cfg: section = pkg_cfg["base"] section["namespace_method"] = section.get("namespace_method", "pkg_util") elif version == 5: pkg_cfg["_pkglts"]["version"] = 6 if "pysetup" in pkg_cfg: section = pkg_cfg["pysetup"] deps = [] for pkg_mng, name in section.get("require", []): if pkg_mng == "none": deps.append(dict(name=name)) else: deps.append(dict(name=name, pkg_mng=pkg_mng)) section["require"] = deps elif version == 6: pkg_cfg["_pkglts"]["version"] = 7 if "sphinx" in pkg_cfg: section = pkg_cfg["sphinx"] section["build_dir"] = section.get("build_dir", "build/sphinx") elif version == 7: pkg_cfg["_pkglts"]["version"] = 8 if "doc" in pkg_cfg: section = pkg_cfg["doc"] section["fmt"] = section.get("fmt", "rst") elif version == 8: pkg_cfg["_pkglts"]["version"] = 9 if "github" in pkg_cfg or "gitlab" in pkg_cfg: LOGGER.info( "\n" "################################################\n" "\n" "\n" "please remove '.gitignore file and regenerate package\n" "\n" "\n" "################################################\n" ) OptionGit("git").update_parameters(pkg_cfg) elif version == 9: pkg_cfg["_pkglts"]["version"] = 10 if "pypi" in pkg_cfg: LOGGER.info( "\n" "################################################\n" "\n" "\n" "please remove '.pypirc file and regenerate package\n" "\n" "\n" "################################################\n" ) # save section mem = dict(pkg_cfg["pypi"]) # get newly defined list of servers OptionPypi("pypi").update_parameters(pkg_cfg) servers = dict(pkg_cfg["pypi"]["servers"]) # update option pkg_cfg["pypi"] = mem pkg_cfg["pypi"]["servers"] = pkg_cfg["pypi"].get("servers", {}) pkg_cfg["pypi"]["servers"].update(servers) elif version == 10: pkg_cfg["_pkglts"]["version"] = 11 if "data" in pkg_cfg: section = pkg_cfg["data"] section["filetype"] = section.get("filetype", [".json", ".ini"]) section["use_ext_dir"] = section.get("use_ext_dir", False) elif version == 11: pkg_cfg["_pkglts"]["version"] = 12 if "pypi" in pkg_cfg: # save servers servers = dict(pkg_cfg["pypi"]["servers"]) # transform dict into list servers_list = [dict(name=name, url=descr["url"]) for name, descr in servers.items()] pkg_cfg["pypi"]["servers"] = servers_list elif version == 12: pkg_cfg["_pkglts"]["version"] = 13 if "base" in pkg_cfg: namespace_method = pkg_cfg["base"].pop("namespace_method") if "src" not in pkg_cfg: OptionSrc("src").update_parameters(pkg_cfg) pkg_cfg["src"]["namespace_method"] = namespace_method if "pysetup" in pkg_cfg: require = pkg_cfg["pysetup"].pop("require") if "reqs" not in pkg_cfg: OptionReqs("reqs").update_parameters(pkg_cfg) pkg_cfg["reqs"]["require"] = require if "sphinx" in pkg_cfg: pkg_cfg["sphinx"]["gallery"] = pkg_cfg["sphinx"].get("gallery", "") elif version == 13: pkg_cfg["_pkglts"]["version"] = 14 if "sphinx" in pkg_cfg: pkg_cfg["sphinx"]["doc_dir"] = pkg_cfg["sphinx"].get("doc_dir", "doc") elif version == 14: pkg_cfg["_pkglts"]["version"] = 15 if "git" in pkg_cfg: try: perm_branches = pkg_cfg["git"]["permanent_branches"] except KeyError: if "gitlab" in pkg_cfg or "github" in pkg_cfg: perm_branches = ["main"] else: perm_branches = [] pkg_cfg["git"]["permanent_branches"] = perm_branches elif version == 15: pkg_cfg["_pkglts"]["version"] = 16 if "conda" in pkg_cfg: try: env_name = pkg_cfg["conda"]["env_name"] except KeyError: env_name = "{{ base.pkgname }}" pkg_cfg["conda"]["env_name"] = env_name elif version == 16: pkg_cfg["_pkglts"]["version"] = 17 if "pysetup" in pkg_cfg: intended_versions = [] for ver in pkg_cfg["pysetup"]["intended_versions"]: if "." in ver: intended_versions.append(ver) else: intended_versions.append(".".join((ver[0], ver[1:]))) pkg_cfg["pysetup"]["intended_versions"] = intended_versions elif version == 17: pkg_cfg["_pkglts"]["version"] = 18 if "pysetup" in pkg_cfg: pkg_cfg["pyproject"] = pkg_cfg.pop("pysetup") return pkg_cfg