meerschaum.config

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

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

250def get_plugin_config(
251        *keys: str,
252        warn: bool = False,
253        **kw: Any
254    ) -> Optional[Any]:
255    """
256    This may only be called from within a Meerschaum plugin.
257    See `meerschaum.config.get_config` for arguments.
258    """
259    from meerschaum.plugins import _get_parent_plugin
260    parent_plugin_name = _get_parent_plugin(2)
261    if parent_plugin_name is None:
262        error(
263            "You may only call `get_plugin_config()` "
264            "from within a Meerschaum plugin."
265        )
266
267    return get_config(
268        *(['plugins', parent_plugin_name] + list(keys)),
269        warn=warn,
270        **kw
271    )

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

274def write_plugin_config(
275        config_dict: Dict[str, Any],
276        **kw : Any
277    ):
278    """
279    Write a plugin's configuration dictionary.
280    """
281    from meerschaum.plugins import _get_parent_plugin
282    parent_plugin_name = _get_parent_plugin(2)
283    if parent_plugin_name is None:
284        error(f"You may only call `get_plugin_config()` from within a Meerschaum plugin.")
285    plugins_cf = get_config('plugins', warn=False)
286    if plugins_cf is None:
287        plugins_cf = {}
288    plugins_cf.update({parent_plugin_name: config_dict})
289    cf = {'plugins' : plugins_cf}
290    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

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

69def set_config(cf: Dict[str, Any]) -> Dict[str, Any]:
70    """
71    Set the configuration dictionary.
72    """
73    global config
74    if not isinstance(cf, dict):
75        from meerschaum.utils.warnings import error
76        error(f"Invalid value for config: {cf}")
77    with _locks['config']:
78        config = cf
79    return config

Set the configuration dictionary.