meerschaum.config

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

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

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

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

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