meerschaum.config

Meerschaum v2.1.7

View Source

  1#! /usr/bin/env python
  2# -*- coding: utf-8 -*-
  3# vim:fenc=utf-8
  4
  5"""
  6Import and update configuration dictionary
  7and if interactive, print the welcome message.
  8"""
  9
 10from __future__ import annotations
 11
 12import os, shutil, sys, pathlib, copy
 13from meerschaum.utils.typing import Any, Dict, Optional, Union
 14from meerschaum.utils.threading import RLock
 15from meerschaum.utils.warnings import warn
 16
 17from meerschaum.config._version import __version__
 18from meerschaum.config._edit import edit_config, write_config
 19from meerschaum.config.static import STATIC_CONFIG
 20
 21from meerschaum.config._paths import (
 22    PERMANENT_PATCH_DIR_PATH,
 23    CONFIG_DIR_PATH,
 24    DEFAULT_CONFIG_DIR_PATH,
 25)
 26from meerschaum.config._patch import (
 27    apply_patch_to_config,
 28)
 29__all__ = ('get_plugin_config', 'write_plugin_config', 'get_config', 'write_config', 'set_config',)
 30__pdoc__ = {'static': False, 'resources': False, 'stack': False, }
 31_locks = {'config': RLock()}
 32
 33### apply config preprocessing (e.g. main to meta)
 34config = {}
 35def _config(
 36        *keys: str, reload: bool = False, substitute: bool = True,
 37        sync_files: bool = True, write_missing: bool = True,
 38    ) -> Dict[str, Any]:
 39    """
 40    Read and process the configuration file.
 41    """
 42    global config
 43    if config is None or reload:
 44        with _locks['config']:
 45            config = {}
 46
 47    if keys and keys[0] not in config:
 48        from meerschaum.config._sync import sync_files as _sync_files
 49        key_config = read_config(
 50            keys = [keys[0]],
 51            substitute = substitute,
 52            write_missing = write_missing,
 53        )
 54        if keys[0] in key_config:
 55            config[keys[0]] = key_config[keys[0]]
 56            if sync_files:
 57                _sync_files(keys=[keys[0] if keys else None])
 58    return config
 59
 60
 61def set_config(cf: Dict[str, Any]) -> Dict[str, Any]:
 62    """
 63    Set the configuration dictionary.
 64    """
 65    global config
 66    if not isinstance(cf, dict):
 67        from meerschaum.utils.warnings import error
 68        error(f"Invalid value for config: {cf}")
 69    with _locks['config']:
 70        config = cf
 71    return config
 72
 73
 74def get_config(
 75        *keys: str,
 76        patch: bool = True,
 77        substitute: bool = True,
 78        sync_files: bool = True,
 79        write_missing: bool = True,
 80        as_tuple: bool = False,
 81        warn: bool = True,
 82        debug: bool = False
 83    ) -> Any:
 84    """
 85    Return the Meerschaum configuration dictionary.
 86    If positional arguments are provided, index by the keys.
 87    Raises a warning if invalid keys are provided.
 88
 89    Parameters
 90    ----------
 91    keys: str:
 92        List of strings to index.
 93
 94    patch: bool, default True
 95        If `True`, patch missing default keys into the config directory.
 96        Defaults to `True`.
 97
 98    sync_files: bool, default True
 99        If `True`, sync files if needed.
100        Defaults to `True`.
101
102    write_missing: bool, default True
103        If `True`, write default values when the main config files are missing.
104        Defaults to `True`.
105
106    substitute: bool, default True
107        If `True`, subsitute 'MRSM{}' values.
108        Defaults to `True`.
109
110    as_tuple: bool, default False
111        If `True`, return a tuple of type (success, value).
112        Defaults to `False`.
113        
114    Returns
115    -------
116    The value in the configuration directory, indexed by the provided keys.
117
118    Examples
119    --------
120    >>> get_config('meerschaum', 'instance')
121    'sql:main'
122    >>> get_config('does', 'not', 'exist')
123    UserWarning: Invalid keys in config: ('does', 'not', 'exist')
124    """
125    import json
126
127    symlinks_key = STATIC_CONFIG['config']['symlinks_key']
128    if debug:
129        from meerschaum.utils.debug import dprint
130        dprint(f"Indexing keys: {keys}", color=False)
131
132    if len(keys) == 0:
133        _rc = _config(substitute=substitute, sync_files=sync_files, write_missing=write_missing)
134        if as_tuple:
135            return True, _rc 
136        return _rc
137    
138    ### Weird threading issues, only import if substitute is True.
139    if substitute:
140        from meerschaum.config._read_config import search_and_substitute_config
141    ### Invalidate the cache if it was read before with substitute=False
142    ### but there still exist substitutions.
143    if (
144        config is not None and substitute and keys[0] != symlinks_key
145        and 'MRSM{' in json.dumps(config.get(keys[0]))
146    ):
147        try:
148            _subbed = search_and_substitute_config({keys[0]: config[keys[0]]})
149        except Exception as e:
150            import traceback
151            traceback.print_exc()
152        config[keys[0]] = _subbed[keys[0]]
153        if symlinks_key in _subbed:
154            if symlinks_key not in config:
155                config[symlinks_key] = {}
156            if keys[0] not in config[symlinks_key]:
157                config[symlinks_key][keys[0]] = {}
158            config[symlinks_key][keys[0]] = apply_patch_to_config(
159                _subbed,
160                config[symlinks_key][keys[0]]
161            )
162
163    from meerschaum.config._sync import sync_files as _sync_files
164    if config is None:
165        _config(*keys, sync_files=sync_files)
166
167    invalid_keys = False
168    if keys[0] not in config and keys[0] != symlinks_key:
169        single_key_config = read_config(
170            keys=[keys[0]], substitute=substitute, write_missing=write_missing
171        )
172        if keys[0] not in single_key_config:
173            invalid_keys = True
174        else:
175            config[keys[0]] = single_key_config.get(keys[0], None)
176            if symlinks_key in single_key_config and keys[0] in single_key_config[symlinks_key]:
177                if symlinks_key not in config:
178                    config[symlinks_key] = {}
179                config[symlinks_key][keys[0]] = single_key_config[symlinks_key][keys[0]]
180
181            if sync_files:
182                _sync_files(keys=[keys[0]])
183
184    c = config
185    if len(keys) > 0:
186        for k in keys:
187            try:
188                c = c[k]
189            except Exception as e:
190                invalid_keys = True
191                break
192        if invalid_keys:
193            ### Check if the keys are in the default configuration.
194            from meerschaum.config._default import default_config
195            in_default = True
196            patched_default_config = (
197                search_and_substitute_config(default_config)
198                if substitute else copy.deepcopy(default_config)
199            )
200            _c = patched_default_config
201            for k in keys:
202                try:
203                    _c = _c[k]
204                except Exception as e:
205                    in_default = False
206            if in_default:
207                c = _c
208                invalid_keys = False
209            warning_msg = f"Invalid keys in config: {keys}"
210            if not in_default:
211                try:
212                    if warn:
213                        from meerschaum.utils.warnings import warn as _warn
214                        _warn(warning_msg, stacklevel=3, color=False)
215                except Exception as e:
216                    if warn:
217                        print(warning_msg)
218                if as_tuple:
219                    return False, None
220                return None
221
222            ### Don't write keys that we haven't yet loaded into memory.
223            not_loaded_keys = [k for k in patched_default_config if k not in config]
224            for k in not_loaded_keys:
225                patched_default_config.pop(k, None)
226
227            set_config(
228                apply_patch_to_config(
229                    patched_default_config,
230                    config,
231                )
232            )
233            if patch and keys[0] != symlinks_key:
234                if write_missing:
235                    write_config(config, debug=debug)
236
237    if as_tuple:
238        return (not invalid_keys), c
239    return c
240
241
242def get_plugin_config(*keys : str, **kw : Any) -> Optional[Any]:
243    """
244    This may only be called from within a Meerschaum plugin.
245    See `meerschaum.config.get_config` for arguments.
246    """
247    from meerschaum.utils.warnings import warn, error
248    from meerschaum.plugins import _get_parent_plugin
249    parent_plugin_name = _get_parent_plugin(2)
250    if parent_plugin_name is None:
251        error(f"You may only call `get_plugin_config()` from within a Meerschaum plugin.")
252    return get_config(*(['plugins', parent_plugin_name] + list(keys)), **kw)
253
254
255def write_plugin_config(
256        config_dict: Dict[str, Any],
257        **kw : Any
258    ):
259    """
260    Write a plugin's configuration dictionary.
261    """
262    from meerschaum.utils.warnings import warn, error
263    from meerschaum.plugins import _get_parent_plugin
264    parent_plugin_name = _get_parent_plugin(2)
265    if parent_plugin_name is None:
266        error(f"You may only call `get_plugin_config()` from within a Meerschaum plugin.")
267    plugins_cf = get_config('plugins', warn=False)
268    if plugins_cf is None:
269        plugins_cf = {}
270    plugins_cf.update({parent_plugin_name: config_dict})
271    cf = {'plugins' : plugins_cf}
272    return write_config(cf, **kw)
273
274
275### This need to be below get_config to avoid a circular import.
276from meerschaum.config._read_config import read_config
277
278### If environment variable MRSM_CONFIG or MRSM_PATCH is set, patch config before anything else.
279from meerschaum.config._environment import apply_environment_patches, apply_environment_uris
280apply_environment_uris()
281apply_environment_patches()
282
283
284from meerschaum.config._paths import PATCH_DIR_PATH, PERMANENT_PATCH_DIR_PATH
285patch_config = None
286if PATCH_DIR_PATH.exists():
287    from meerschaum.utils.yaml import yaml, _yaml
288    if _yaml is not None:
289        patch_config = read_config(directory=PATCH_DIR_PATH)
290
291permanent_patch_config = None
292if PERMANENT_PATCH_DIR_PATH.exists():
293    from meerschaum.utils.yaml import yaml, _yaml
294    if _yaml is not None:
295        permanent_patch_config = read_config(directory=PERMANENT_PATCH_DIR_PATH)
296### If patches exist, apply to config.
297if patch_config is not None:
298    from meerschaum.config._paths import PATCH_DIR_PATH
299    set_config(apply_patch_to_config(_config(), patch_config))
300    if PATCH_DIR_PATH.exists():
301        shutil.rmtree(PATCH_DIR_PATH)
302
303### if permanent_patch.yaml exists, apply patch to config, write config, and delete patch
304if permanent_patch_config is not None and PERMANENT_PATCH_DIR_PATH.exists():
305    print(
306        "Found permanent patch configuration. " +
307        "Updating main config and deleting permanent patch..."
308    )
309    set_config(apply_patch_to_config(_config(), permanent_patch_config))
310    write_config(_config())
311    permanent_patch_config = None
312    if PERMANENT_PATCH_DIR_PATH.exists():
313        shutil.rmtree(PERMANENT_PATCH_DIR_PATH)
314    if DEFAULT_CONFIG_DIR_PATH.exists():
315        shutil.rmtree(DEFAULT_CONFIG_DIR_PATH)
316
317
318### Make sure readline is available for the portable version.
319environment_runtime = STATIC_CONFIG['environment']['runtime']
320if environment_runtime in os.environ:
321    if os.environ[environment_runtime] == 'portable':
322        from meerschaum.utils.packages import ensure_readline
323        from meerschaum.config._paths import PORTABLE_CHECK_READLINE_PATH
324        if not PORTABLE_CHECK_READLINE_PATH.exists():
325            ensure_readline()
326            PORTABLE_CHECK_READLINE_PATH.touch()
327
328
329### If interactive REPL, print welcome header.
330__doc__ = f"Meerschaum v{__version__}"
331try:
332    interactive = False
333    if sys.ps1:
334        interactive = True
335except AttributeError:
336    interactive = False
337if interactive:
338    msg = __doc__
339    print(msg, file=sys.stderr)

def get_plugin_config(*keys: str, **kw: Any) -> Optional[Any]: View Source

243def get_plugin_config(*keys : str, **kw : Any) -> Optional[Any]:
244    """
245    This may only be called from within a Meerschaum plugin.
246    See `meerschaum.config.get_config` for arguments.
247    """
248    from meerschaum.utils.warnings import warn, error
249    from meerschaum.plugins import _get_parent_plugin
250    parent_plugin_name = _get_parent_plugin(2)
251    if parent_plugin_name is None:
252        error(f"You may only call `get_plugin_config()` from within a Meerschaum plugin.")
253    return get_config(*(['plugins', parent_plugin_name] + list(keys)), **kw)

This may only be called from within a Meerschaum plugin. See get_config for arguments.

def write_plugin_config(config_dict: Dict[str, Any], **kw: Any): View Source

256def write_plugin_config(
257        config_dict: Dict[str, Any],
258        **kw : Any
259    ):
260    """
261    Write a plugin's configuration dictionary.
262    """
263    from meerschaum.utils.warnings import warn, error
264    from meerschaum.plugins import _get_parent_plugin
265    parent_plugin_name = _get_parent_plugin(2)
266    if parent_plugin_name is None:
267        error(f"You may only call `get_plugin_config()` from within a Meerschaum plugin.")
268    plugins_cf = get_config('plugins', warn=False)
269    if plugins_cf is None:
270        plugins_cf = {}
271    plugins_cf.update({parent_plugin_name: config_dict})
272    cf = {'plugins' : plugins_cf}
273    return write_config(cf, **kw)

Write a plugin's configuration dictionary.

def get_config( *keys: str, patch: bool = True, substitute: bool = True, sync_files: bool = True, write_missing: bool = True, as_tuple: bool = False, warn: bool = True, debug: bool = False) -> Any: View Source

 75def get_config(
 76        *keys: str,
 77        patch: bool = True,
 78        substitute: bool = True,
 79        sync_files: bool = True,
 80        write_missing: bool = True,
 81        as_tuple: bool = False,
 82        warn: bool = True,
 83        debug: bool = False
 84    ) -> Any:
 85    """
 86    Return the Meerschaum configuration dictionary.
 87    If positional arguments are provided, index by the keys.
 88    Raises a warning if invalid keys are provided.
 89
 90    Parameters
 91    ----------
 92    keys: str:
 93        List of strings to index.
 94
 95    patch: bool, default True
 96        If `True`, patch missing default keys into the config directory.
 97        Defaults to `True`.
 98
 99    sync_files: bool, default True
100        If `True`, sync files if needed.
101        Defaults to `True`.
102
103    write_missing: bool, default True
104        If `True`, write default values when the main config files are missing.
105        Defaults to `True`.
106
107    substitute: bool, default True
108        If `True`, subsitute 'MRSM{}' values.
109        Defaults to `True`.
110
111    as_tuple: bool, default False
112        If `True`, return a tuple of type (success, value).
113        Defaults to `False`.
114        
115    Returns
116    -------
117    The value in the configuration directory, indexed by the provided keys.
118
119    Examples
120    --------
121    >>> get_config('meerschaum', 'instance')
122    'sql:main'
123    >>> get_config('does', 'not', 'exist')
124    UserWarning: Invalid keys in config: ('does', 'not', 'exist')
125    """
126    import json
127
128    symlinks_key = STATIC_CONFIG['config']['symlinks_key']
129    if debug:
130        from meerschaum.utils.debug import dprint
131        dprint(f"Indexing keys: {keys}", color=False)
132
133    if len(keys) == 0:
134        _rc = _config(substitute=substitute, sync_files=sync_files, write_missing=write_missing)
135        if as_tuple:
136            return True, _rc 
137        return _rc
138    
139    ### Weird threading issues, only import if substitute is True.
140    if substitute:
141        from meerschaum.config._read_config import search_and_substitute_config
142    ### Invalidate the cache if it was read before with substitute=False
143    ### but there still exist substitutions.
144    if (
145        config is not None and substitute and keys[0] != symlinks_key
146        and 'MRSM{' in json.dumps(config.get(keys[0]))
147    ):
148        try:
149            _subbed = search_and_substitute_config({keys[0]: config[keys[0]]})
150        except Exception as e:
151            import traceback
152            traceback.print_exc()
153        config[keys[0]] = _subbed[keys[0]]
154        if symlinks_key in _subbed:
155            if symlinks_key not in config:
156                config[symlinks_key] = {}
157            if keys[0] not in config[symlinks_key]:
158                config[symlinks_key][keys[0]] = {}
159            config[symlinks_key][keys[0]] = apply_patch_to_config(
160                _subbed,
161                config[symlinks_key][keys[0]]
162            )
163
164    from meerschaum.config._sync import sync_files as _sync_files
165    if config is None:
166        _config(*keys, sync_files=sync_files)
167
168    invalid_keys = False
169    if keys[0] not in config and keys[0] != symlinks_key:
170        single_key_config = read_config(
171            keys=[keys[0]], substitute=substitute, write_missing=write_missing
172        )
173        if keys[0] not in single_key_config:
174            invalid_keys = True
175        else:
176            config[keys[0]] = single_key_config.get(keys[0], None)
177            if symlinks_key in single_key_config and keys[0] in single_key_config[symlinks_key]:
178                if symlinks_key not in config:
179                    config[symlinks_key] = {}
180                config[symlinks_key][keys[0]] = single_key_config[symlinks_key][keys[0]]
181
182            if sync_files:
183                _sync_files(keys=[keys[0]])
184
185    c = config
186    if len(keys) > 0:
187        for k in keys:
188            try:
189                c = c[k]
190            except Exception as e:
191                invalid_keys = True
192                break
193        if invalid_keys:
194            ### Check if the keys are in the default configuration.
195            from meerschaum.config._default import default_config
196            in_default = True
197            patched_default_config = (
198                search_and_substitute_config(default_config)
199                if substitute else copy.deepcopy(default_config)
200            )
201            _c = patched_default_config
202            for k in keys:
203                try:
204                    _c = _c[k]
205                except Exception as e:
206                    in_default = False
207            if in_default:
208                c = _c
209                invalid_keys = False
210            warning_msg = f"Invalid keys in config: {keys}"
211            if not in_default:
212                try:
213                    if warn:
214                        from meerschaum.utils.warnings import warn as _warn
215                        _warn(warning_msg, stacklevel=3, color=False)
216                except Exception as e:
217                    if warn:
218                        print(warning_msg)
219                if as_tuple:
220                    return False, None
221                return None
222
223            ### Don't write keys that we haven't yet loaded into memory.
224            not_loaded_keys = [k for k in patched_default_config if k not in config]
225            for k in not_loaded_keys:
226                patched_default_config.pop(k, None)
227
228            set_config(
229                apply_patch_to_config(
230                    patched_default_config,
231                    config,
232                )
233            )
234            if patch and keys[0] != symlinks_key:
235                if write_missing:
236                    write_config(config, debug=debug)
237
238    if as_tuple:
239        return (not invalid_keys), c
240    return c

Return the Meerschaum configuration dictionary. If positional arguments are provided, index by the keys. Raises a warning if invalid keys are provided.

Parameters

keys (str:): List of strings to index.
patch (bool, default True): If True, patch missing default keys into the config directory. Defaults to True.
sync_files (bool, default True): If True, sync files if needed. Defaults to True.
write_missing (bool, default True): If True, write default values when the main config files are missing. Defaults to True.
substitute (bool, default True): If True, subsitute 'MRSM{}' values. Defaults to True.
as_tuple (bool, default False): If True, return a tuple of type (success, value). Defaults to False.

Returns

The value in the configuration directory, indexed by the provided keys.

Examples

>>> get_config('meerschaum', 'instance')
'sql:main'
>>> get_config('does', 'not', 'exist')
UserWarning: Invalid keys in config: ('does', 'not', 'exist')

def write_config( config_dict: Optional[Dict[str, Any]] = None, directory: Union[str, pathlib.Path, NoneType] = None, debug: bool = False, **kw: Any) -> bool: View Source

 43def write_config(
 44        config_dict: Optional[Dict[str, Any]] = None,
 45        directory: Union[str, pathlib.Path, None] = None,
 46        debug: bool = False,
 47        **kw : Any
 48    ) -> bool:
 49    """Write YAML and JSON files to the configuration directory.
 50
 51    Parameters
 52    ----------
 53    config_dict: Optional[Dict[str, Any]], default None
 54        A dictionary of keys to dictionaries of configuration.
 55        Each key corresponds to a .yaml or .json config file.
 56        Writing config to a directory with different keys
 57        does not affect existing keys in that directory.
 58        If not provided, use the currently loaded config dictionary.
 59
 60    directory: Union[str, pathlib.Path, None], default None
 61        The directory to which the keys are written.
 62        If not provided, use the default config path (`~/.config/meerschaum/config/`).
 63
 64    Returns
 65    -------
 66    A bool indicating success.
 67
 68    """
 69    if directory is None:
 70        from meerschaum.config._paths import CONFIG_DIR_PATH
 71        directory = CONFIG_DIR_PATH
 72    from meerschaum.config.static import STATIC_CONFIG
 73    from meerschaum.config._default import default_header_comment
 74    from meerschaum.config._patch import apply_patch_to_config
 75    from meerschaum.config._read_config import get_keyfile_path
 76    from meerschaum.utils.debug import dprint
 77    from meerschaum.utils.yaml import yaml
 78    from meerschaum.utils.misc import filter_keywords
 79    import json, os
 80    if config_dict is None:
 81        from meerschaum.config import _config
 82        cf = _config()
 83        config_dict = cf
 84
 85    default_filetype = STATIC_CONFIG['config']['default_filetype']
 86    filetype_dumpers = {
 87        'yml' : yaml.dump,
 88        'yaml' : yaml.dump,
 89        'json' : json.dump,
 90    }
 91
 92    symlinks_key = STATIC_CONFIG['config']['symlinks_key']
 93    symlinks = config_dict.pop(symlinks_key) if symlinks_key in config_dict else {}
 94    config_dict = apply_patch_to_config(config_dict, symlinks)
 95
 96    def determine_filetype(k, v):
 97        if k == 'meerschaum':
 98            return 'yaml'
 99        if isinstance(v, dict) and 'filetype' in v:
100            return v['filetype']
101        path = get_keyfile_path(k, create_new=False, directory=directory)
102        if path is None:
103            return default_filetype
104        filetype = path.suffix[1:]
105        if not isinstance(filetype, str) or filetype not in filetype_dumpers:
106            print(f"Invalid filetype '{filetype}' for '{k}'. Assuming {default_filetype}...")
107            filetype = default_filetype
108        return filetype
109
110    for k, v in config_dict.items():
111        filetype = determine_filetype(k, v)        
112        filename = str(k) + '.' + str(filetype)
113        filepath = os.path.join(directory, filename)
114        pathlib.Path(filepath).parent.mkdir(exist_ok=True)
115        with open(filepath, 'w+') as f:
116            try:
117                if k == 'meerschaum':
118                    f.write(default_header_comment)
119                filetype_dumpers[filetype](
120                    v, f,
121                    **filter_keywords(
122                        filetype_dumpers[filetype],
123                        sort_keys = False,
124                        indent = 2
125                    )
126                )
127                success = True
128            except Exception as e:
129                success = False
130                print(f"FAILED TO WRITE!")
131                print(e)
132                print(filter_keywords(
133                    filetype_dumpers[filetype],
134                    sort_keys=False,
135                    indent = 2
136                ))
137
138            if not success:
139                try:
140                    if os.path.exists(filepath):
141                        os.remove(filepath)
142                except Exception as e:
143                    print(f"Failed to write '{k}'")
144                return False
145
146    return True

Write YAML and JSON files to the configuration directory.

Parameters

config_dict (Optional[Dict[str, Any]], default None): A dictionary of keys to dictionaries of configuration. Each key corresponds to a .yaml or .json config file. Writing config to a directory with different keys does not affect existing keys in that directory. If not provided, use the currently loaded config dictionary.
directory (Union[str, pathlib.Path, None], default None): The directory to which the keys are written. If not provided, use the default config path (~/.config/meerschaum/config/).

Returns

A bool indicating success.

def set_config(cf: Dict[str, Any]) -> Dict[str, Any]: View Source

62def set_config(cf: Dict[str, Any]) -> Dict[str, Any]:
63    """
64    Set the configuration dictionary.
65    """
66    global config
67    if not isinstance(cf, dict):
68        from meerschaum.utils.warnings import error
69        error(f"Invalid value for config: {cf}")
70    with _locks['config']:
71        config = cf
72    return config

Set the configuration dictionary.