meerschaum.config

283def get_plugin_config(
284    *keys: str,
285    warn: bool = False,
286    **kw: Any
287) -> Optional[Any]:
288    """
289    This may only be called from within a Meerschaum plugin.
290    See `meerschaum.config.get_config` for arguments.
291    """
292    from meerschaum.utils.warnings import error
293    from meerschaum.plugins import _get_parent_plugin
294    parent_plugin_name = _get_parent_plugin(2)
295    if parent_plugin_name is None:
296        error(
297            "You may only call `get_plugin_config()` "
298            "from within a Meerschaum plugin."
299        )
300
301    return get_config(
302        *(['plugins', parent_plugin_name] + list(keys)),
303        warn=warn,
304        **kw
305    )

308def write_plugin_config(
309    config_dict: Dict[str, Any],
310    **kw: Any
311):
312    """
313    Write a plugin's configuration dictionary.
314    """
315    from meerschaum.utils.warnings import error
316    from meerschaum.plugins import _get_parent_plugin
317    parent_plugin_name = _get_parent_plugin(2)
318    if parent_plugin_name is None:
319        error("You may only call `get_plugin_config()` from within a Meerschaum plugin.")
320    plugins_cf = get_config('plugins', warn=False)
321    if plugins_cf is None:
322        plugins_cf = {}
323    plugins_cf.update({parent_plugin_name: config_dict})
324    cf = {'plugins' : plugins_cf}
325    return write_config(cf, **kw)

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

 78def write_config(
 79    config_dict: Optional[Dict[str, Any]] = None,
 80    directory: Union[str, pathlib.Path, None] = None,
 81    debug: bool = False,
 82    **kw: Any
 83) -> bool:
 84    """Write YAML and JSON files to the configuration directory.
 85
 86    Parameters
 87    ----------
 88    config_dict: Optional[Dict[str, Any]], default None
 89        A dictionary of keys to dictionaries of configuration.
 90        Each key corresponds to a .yaml or .json config file.
 91        Writing config to a directory with different keys
 92        does not affect existing keys in that directory.
 93        If not provided, use the currently loaded config dictionary.
 94
 95    directory: Union[str, pathlib.Path, None], default None
 96        The directory to which the keys are written.
 97        If not provided, use the default config path (`~/.config/meerschaum/config/`).
 98
 99    Returns
100    -------
101    A bool indicating success.
102
103    """
104    if directory is None:
105        import meerschaum.config.paths as paths
106        directory = paths.CONFIG_DIR_PATH
107
108    from meerschaum.config import _allow_write_missing
109    from meerschaum._internal.static import STATIC_CONFIG
110    from meerschaum.config._default import default_header_comment
111    from meerschaum.config._read_config import get_keyfile_path, revert_symlinks_config
112    from meerschaum.utils.yaml import yaml
113    from meerschaum.utils.misc import filter_keywords
114    import json
115    import os
116    if config_dict is None:
117        from meerschaum.config import _config
118        cf = _config(allow_replaced=False)
119        config_dict = cf
120
121    if not _allow_write_missing:
122        return False
123
124    default_filetype = STATIC_CONFIG['config']['default_filetype']
125    filetype_dumpers = {
126        'yml' : yaml.dump,
127        'yaml' : yaml.dump,
128        'json' : json.dump,
129    }
130
131    config_dict = revert_symlinks_config(config_dict)
132
133    def determine_filetype(k, v):
134        if k == 'meerschaum':
135            return 'yaml'
136        if isinstance(v, dict) and 'filetype' in v:
137            return v['filetype']
138        path = get_keyfile_path(k, create_new=False, directory=directory)
139        if path is None:
140            return default_filetype
141        filetype = path.suffix[1:]
142        if not isinstance(filetype, str) or filetype not in filetype_dumpers:
143            print(f"Invalid filetype '{filetype}' for '{k}'. Assuming {default_filetype}...")
144            filetype = default_filetype
145        return filetype
146
147    for k, v in config_dict.items():
148        filetype = determine_filetype(k, v)        
149        filename = str(k) + '.' + str(filetype)
150        filepath = os.path.join(directory, filename)
151        pathlib.Path(filepath).parent.mkdir(exist_ok=True)
152        with open(filepath, 'w+') as f:
153            try:
154                if k == 'meerschaum':
155                    f.write(default_header_comment)
156                filetype_dumpers[filetype](
157                    v, f,
158                    **filter_keywords(
159                        filetype_dumpers[filetype],
160                        sort_keys = False,
161                        indent = 2
162                    )
163                )
164                success = True
165            except Exception as e:
166                success = False
167                print("FAILED TO WRITE!")
168                print(e)
169                print(filter_keywords(
170                    filetype_dumpers[filetype],
171                    sort_keys=False,
172                    indent = 2
173                ))
174
175            if not success:
176                try:
177                    if os.path.exists(filepath):
178                        os.remove(filepath)
179                except Exception:
180                    print(f"Failed to write '{k}'")
181                return False
182
183    return True

16def edit_config(
17    keys: Optional[List[str]] = None,
18    params: Optional[Dict[str, Any]] = None,
19    debug: bool = False,
20    **kw: Any
21) -> SuccessTuple:
22    """Edit the configuration files."""
23    import meerschaum.config.paths as paths
24    from meerschaum.config import get_config, config
25    from meerschaum.config._read_config import get_keyfile_path, read_config, revert_symlinks_config
26    from meerschaum._internal.static import STATIC_CONFIG
27    from meerschaum.utils.packages import reload_meerschaum
28    from meerschaum.utils.misc import edit_file
29    from meerschaum.utils.warnings import warn
30    from meerschaum.utils.prompt import prompt
31
32    if keys is None:
33        keys = []
34
35    symlinks_key = STATIC_CONFIG['config']['symlinks_key']
36    def _edit_key(key: str):
37        new_key_config = None
38        while True:
39            ### If defined in default, create the config file.
40            symlinks_key_config = config.get(symlinks_key, {}).get(key, {})
41            key_config = revert_symlinks_config({
42                key: config.pop(key, {}),
43                symlinks_key: {key: symlinks_key_config},
44            })
45            keyfile_path = get_keyfile_path(key, create_new=True)
46            get_config(key, write_missing=True, warn=False)
47
48            edit_file(get_keyfile_path(key, create_new=True))
49
50            ### TODO: verify that the file is valid. Retry if not.
51            try:
52                new_key_config = read_config(
53                    paths.CONFIG_DIR_PATH,
54                    [key],
55                    write_missing=False,
56                    raise_parsing_errors=True,
57                )
58            except Exception:
59                if key_config:
60                    config[key] = key_config
61                warn(f"Could not parse key '{key}'.", stack=False)
62                _ = prompt(f"Press [Enter] to edit '{keyfile_path}', [CTRL+C] to exit.")
63                continue
64
65            if new_key_config:
66                break
67
68    try:
69        for k in keys:
70            _edit_key(k)
71    except KeyboardInterrupt:
72        return False, ""
73
74    reload_meerschaum(debug=debug)
75    return (True, "Success")

 98def set_config(cf: Dict[str, Any]) -> Dict[str, Any]:
 99    """
100    Set the configuration dictionary.
101    """
102    global config
103    if not isinstance(cf, dict):
104        from meerschaum.utils.warnings import error
105        error(f"Invalid value for config: {cf}")
106    with _locks['config']:
107        config = cf
108    return config

328@contextlib.contextmanager
329def replace_config(config_: Union[Dict[str, Any], None]):
330    """
331    Temporarily override the Meerschaum config dictionary.
332
333    Parameters
334    ----------
335    config_: Dict[str, Any]
336        The new config dictionary to temporarily replace the canonical `config`.
337    """
338    if config_ is None:
339        try:
340            yield
341        except Exception:
342            pass
343        return
344
345    global _backup_config, _allow_write_missing
346
347    _backup_config = _config()
348    _allow_write_missing = False
349    set_config(config_)
350
351    try:
352        yield
353    finally:
354        set_config(_backup_config)
355        _allow_write_missing = True

216def search_and_substitute_config(
217    config: Dict[str, Any],
218    leading_key: str = "MRSM",
219    delimiter: str = ":",
220    begin_key: str = "{",
221    end_key: str = "}",
222    literal_key: str = '!',
223    keep_symlinks: bool = True,
224) -> Dict[str, Any]:
225    """
226    Search the config for Meerschaum substitution syntax and substite with value of keys.
227
228    Parameters
229    ----------
230    config: Dict[str, Any]
231        The Meerschaum configuration dictionary to search through.
232
233    leading_key: str, default 'MRSM'
234        The string with which to start the search.
235
236    begin_key: str, default '{'
237        The string to start the keys list.
238
239    end_key: str, default '}'
240        The string to end the keys list.
241
242    literal_key: str, default '!'
243        The string to force an literal interpretation of a value.
244        When the string is isolated, a literal interpreation is assumed and the surrounding
245        quotes are replaced.
246        
247        E.g. Suppose a:b:c produces a dictionary {'d': 1}.
248        - 'MRSM{a:b:c}'    => {'d': 1}        : isolated
249        - ' MRSM{a:b:c} '  => ' "{\'d\': 1}"' : not isolated
250        - ' MRSM{!a:b:c} ' => ' {"d": 1}'     : literal
251
252    keep_symlinks: bool, default True
253        If True, include the symlinks under the top-level key '_symlinks' (never written to a file).
254        Defaults to True.
255        
256        Example:
257
258        ```
259        MRSM{meerschaum:connectors:main:host} => cf['meerschaum']['connectors']['main']['host']
260        ``` 
261
262    Returns
263    -------
264    The configuration dictionary with `MRSM{}` symlinks replaced with
265    the values from the current configuration.
266    """
267    import re
268    import json
269    from meerschaum.config import get_config
270    needle = leading_key + begin_key
271
272    _links = []
273    if keep_symlinks:
274        def _find_symlinks(d, _keys: Optional[List[str]] = None):
275            if _keys is None:
276                _keys = []
277            if not isinstance(d, dict):
278                return
279            for k, v in d.items():
280                if isinstance(v, dict):
281                    _find_symlinks(v, _keys + [k])
282                elif isinstance(v, str) and needle in v:
283                    _links.append((_keys + [k], v))
284        _find_symlinks(config)
285
286    haystack = json.dumps(config, separators=(',', ':'))
287    if needle not in haystack:
288        parsed_config = config
289    else:
290        patterns = {}
291        isolated_patterns = {}
292        literal_patterns = {}
293        memo = {}
294
295        pattern_re = re.compile(
296            re.escape(needle) + r'(?P<pattern_keys>.*?)' + re.escape(end_key)
297        )
298
299        for match in pattern_re.finditer(haystack):
300            pattern = match.group(0)
301            pattern_keys_raw = match.group('pattern_keys')
302            pattern_keys = pattern_keys_raw.split(delimiter)
303
304            start, end = match.span()
305            prior = haystack[start - 1] if start > 0 else None
306            after = haystack[end] if end < len(haystack) else None
307
308            force_literal = False
309            keys = [k for k in pattern_keys]
310            if keys and str(keys[0]).startswith(literal_key):
311                keys[0] = str(keys[0])[len(literal_key):]
312                force_literal = True
313            if len(keys) == 1 and keys[0] == '':
314                keys = []
315
316            cache_key = tuple(keys)
317            if cache_key in memo:
318                valid, value = memo[cache_key]
319            else:
320                try:
321                    valid, value = get_config(
322                        *keys,
323                        substitute=False,
324                        as_tuple=True,
325                        write_missing=False,
326                        sync_files=False,
327                    )
328                except Exception:
329                    valid, value = False, None
330                memo[cache_key] = (valid, value)
331
332            if not valid:
333                continue
334
335            patterns[pattern] = value
336            isolated_patterns[pattern] = (prior == '"' and after == '"')
337            literal_patterns[pattern] = force_literal
338
339        for pattern, value in patterns.items():
340            if isolated_patterns[pattern]:
341                haystack = haystack.replace(
342                    json.dumps(pattern),
343                    json.dumps(value),
344                )
345            elif literal_patterns[pattern]:
346                haystack = haystack.replace(
347                    pattern,
348                    (
349                        json.dumps(value)
350                        .replace("\\", "\\\\")
351                        .replace('"', '\\"')
352                        .replace("'", "\\'")
353                    )
354                )
355            else:
356                haystack = haystack.replace(pattern, str(value))
357
358        parsed_config = json.loads(haystack) or {}
359
360    symlinks = {}
361    if keep_symlinks:
362        for _keys, _pattern in _links:
363            s = symlinks
364            for k in _keys[:-1]:
365                if k not in s:
366                    s[k] = {}
367                s = s[k]
368            s[_keys[-1]] = _pattern
369
370        from meerschaum.config._patch import apply_patch_to_config
371        symlinks_key = STATIC_CONFIG['config']['symlinks_key']
372        if symlinks:
373            if symlinks_key not in parsed_config:
374                parsed_config[symlinks_key] = symlinks
375            else:
376                parsed_config[symlinks_key] = apply_patch_to_config(
377                    parsed_config[symlinks_key],
378                    symlinks,
379                    warn=False,
380                )
381
382    return parsed_config

385def revert_symlinks_config(config: Dict[str, Any]) -> Dict[str, Any]:
386    """
387    Given a configuration dictionary, re-apply the values from the
388    accompanying `_symlinks` dictionary.
389
390    Parameters
391    ----------
392    config: Dict[str, Any]
393        The configuration dictionary containing a `_symlinks` dictionary.
394
395    Returns
396    -------
397    A configuration dictionary with `_symlinks` re-applied.
398    """
399    import copy
400    from meerschaum._internal.static import STATIC_CONFIG
401
402    symlinks_key = STATIC_CONFIG['config']['symlinks_key']
403    if symlinks_key not in config:
404        return config
405
406    reverted_config = copy.deepcopy(config)
407    symlinks_config = reverted_config.pop(symlinks_key)
408
409    def deep_patch(target_dict, patch_dict):
410        for key, value in patch_dict.items():
411            if isinstance(value, dict) and key in target_dict and isinstance(target_dict[key], dict):
412                deep_patch(target_dict[key], value)
413            else:
414                target_dict[key] = value
415
416    deep_patch(reverted_config, symlinks_config)
417    return reverted_config

420def get_possible_keys() -> List[str]:
421    """
422    Return a list of possible top-level keys.
423    """
424    import os
425    import meerschaum.config.paths as paths
426    from meerschaum.config._default import default_config
427    keys = set()
428    for key in default_config:
429        keys.add(key)
430    for filename in os.listdir(paths.CONFIG_DIR_PATH):
431        keys.add('.'.join(filename.split('.')[:-1]))
432    return sorted(list(keys))

435def get_keyfile_path(
436    key: str,
437    create_new: bool = False,
438    directory: Union[pathlib.Path, str, None] = None,
439) -> Union[pathlib.Path, None]:
440    """Determine a key's file path."""
441    import os
442    import pathlib
443    if directory is None:
444        import meerschaum.config.paths as paths
445        directory = paths.CONFIG_DIR_PATH
446
447    try:
448        return pathlib.Path(
449            os.path.join(
450                directory,
451                read_config(
452                    keys=[key],
453                    with_filenames=True,
454                    write_missing=False,
455                    substitute=False,
456                )[1][0]
457            )
458        )
459    except IndexError:
460        if create_new:
461            default_filetype = STATIC_CONFIG['config']['default_filetype']
462            return pathlib.Path(os.path.join(directory, key + '.' + default_filetype))
463        return None