meerschaum.config

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

315def write_plugin_config(
316    config_dict: Dict[str, Any],
317    **kw: Any
318):
319    """
320    Write a plugin's configuration dictionary.
321    """
322    from meerschaum.utils.warnings import error
323    from meerschaum.plugins import _get_parent_plugin
324    parent_plugin_name = _get_parent_plugin(2)
325    if parent_plugin_name is None:
326        error("You may only call `get_plugin_config()` from within a Meerschaum plugin.")
327    plugins_cf = get_config('plugins', warn=False)
328    if plugins_cf is None:
329        plugins_cf = {}
330    plugins_cf.update({parent_plugin_name: config_dict})
331    cf = {'plugins' : plugins_cf}
332    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 and write_missing:
276                ### Only persist defaults when the key's file is genuinely absent.
277                ### Never overwrite an existing file (e.g. one that failed to parse) ─
278                ### doing so would clobber the user's config with default values.
279                ### Brand-new config files are still created by `read_config()`.
280                from meerschaum.config._read_config import get_keyfile_path
281                keyfile_exists = get_keyfile_path(keys[0], create_new=False) is not None
282                if not keyfile_exists:
283                    write_config(config, debug=debug)
284
285    if as_tuple:
286        return (not invalid_keys), c
287    return c

 99def write_config(
100    config_dict: Optional[Dict[str, Any]] = None,
101    directory: Union[str, pathlib.Path, None] = None,
102    debug: bool = False,
103    **kw: Any
104) -> bool:
105    """Write YAML and JSON files to the configuration directory.
106
107    Parameters
108    ----------
109    config_dict: Optional[Dict[str, Any]], default None
110        A dictionary of keys to dictionaries of configuration.
111        Each key corresponds to a .yaml or .json config file.
112        Writing config to a directory with different keys
113        does not affect existing keys in that directory.
114        If not provided, use the currently loaded config dictionary.
115
116    directory: Union[str, pathlib.Path, None], default None
117        The directory to which the keys are written.
118        If not provided, use the default config path (`~/.config/meerschaum/config/`).
119
120    Returns
121    -------
122    A bool indicating success.
123
124    """
125    if directory is None:
126        import meerschaum.config.paths as paths
127        directory = paths.CONFIG_DIR_PATH
128
129    from meerschaum.config import _allow_write_missing
130    from meerschaum._internal.static import STATIC_CONFIG
131    from meerschaum.config._default import default_header_comment
132    from meerschaum.config._read_config import get_keyfile_path, revert_symlinks_config
133    from meerschaum.utils.yaml import yaml
134    from meerschaum.utils.misc import filter_keywords
135    import json
136    import os
137    if config_dict is None:
138        from meerschaum.config import _config
139        cf = _config(allow_replaced=False)
140        config_dict = cf
141
142    if not _allow_write_missing:
143        return False
144
145    default_filetype = STATIC_CONFIG['config']['default_filetype']
146    filetype_dumpers = {
147        'yml' : yaml.dump,
148        'yaml' : yaml.dump,
149        'json' : json.dump,
150    }
151
152    config_dict = revert_symlinks_config(config_dict)
153
154    def determine_filetype(k, v):
155        if k == 'meerschaum':
156            return 'yaml'
157        if isinstance(v, dict) and 'filetype' in v:
158            return v['filetype']
159        path = get_keyfile_path(k, create_new=False, directory=directory)
160        if path is None:
161            return default_filetype
162        filetype = path.suffix[1:]
163        if not isinstance(filetype, str) or filetype not in filetype_dumpers:
164            print(f"Invalid filetype '{filetype}' for '{k}'. Assuming {default_filetype}...")
165            filetype = default_filetype
166        return filetype
167
168    for k, v in config_dict.items():
169        filetype = determine_filetype(k, v)        
170        filename = str(k) + '.' + str(filetype)
171        filepath = os.path.join(directory, filename)
172        pathlib.Path(filepath).parent.mkdir(exist_ok=True)
173        with open(filepath, 'w+') as f:
174            try:
175                if k == 'meerschaum':
176                    f.write(default_header_comment)
177                filetype_dumpers[filetype](
178                    v, f,
179                    **filter_keywords(
180                        filetype_dumpers[filetype],
181                        sort_keys = False,
182                        indent = 2
183                    )
184                )
185                success = True
186            except Exception as e:
187                success = False
188                print("FAILED TO WRITE!")
189                print(e)
190                print(filter_keywords(
191                    filetype_dumpers[filetype],
192                    sort_keys=False,
193                    indent = 2
194                ))
195
196            if not success:
197                try:
198                    if os.path.exists(filepath):
199                        os.remove(filepath)
200                except Exception:
201                    print(f"Failed to write '{k}'")
202                return False
203
204    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.config._backup import backup_config_key, restore_config_backup
27    from meerschaum._internal.static import STATIC_CONFIG
28    from meerschaum.utils.packages import reload_meerschaum
29    from meerschaum.utils.misc import edit_file
30    from meerschaum.utils.warnings import warn, info
31    from meerschaum.utils.prompt import prompt, yes_no
32
33    if keys is None:
34        keys = []
35
36    symlinks_key = STATIC_CONFIG['config']['symlinks_key']
37    def _edit_key(key: str):
38        new_key_config = None
39        ### Snapshot the pristine file before any edits or default-merging so it
40        ### can be rolled back if the edit corrupts it.
41        backup_path = backup_config_key(key, debug=debug)
42        while True:
43            ### If defined in default, create the config file.
44            symlinks_key_config = config.get(symlinks_key, {}).get(key, {})
45            key_config = revert_symlinks_config({
46                key: config.pop(key, {}),
47                symlinks_key: {key: symlinks_key_config},
48            })
49            keyfile_path = get_keyfile_path(key, create_new=True)
50            get_config(key, write_missing=True, warn=False)
51
52            edit_file(get_keyfile_path(key, create_new=True))
53
54            ### TODO: verify that the file is valid. Retry if not.
55            try:
56                new_key_config = read_config(
57                    paths.CONFIG_DIR_PATH,
58                    [key],
59                    write_missing=False,
60                    raise_parsing_errors=True,
61                )
62            except Exception:
63                if key_config:
64                    config[key] = key_config
65                warn(f"Could not parse key '{key}'.", stack=False)
66                if backup_path is not None and yes_no(
67                    f"Restore '{key}' from the backup taken before editing?",
68                    default='n',
69                ):
70                    restore_success, restore_msg = restore_config_backup(
71                        key, backup_path=backup_path, reload=False, debug=debug,
72                    )
73                    if restore_success:
74                        info(restore_msg)
75                        new_key_config = read_config(
76                            paths.CONFIG_DIR_PATH,
77                            [key],
78                            write_missing=False,
79                            raise_parsing_errors=False,
80                        )
81                        break
82                    warn(restore_msg, stack=False)
83                _ = prompt(f"Press [Enter] to edit '{keyfile_path}', [CTRL+C] to exit.")
84                continue
85
86            if new_key_config:
87                break
88
89    try:
90        for k in keys:
91            _edit_key(k)
92    except KeyboardInterrupt:
93        return False, ""
94
95    reload_meerschaum(debug=debug)
96    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

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

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

396def revert_symlinks_config(config: Dict[str, Any]) -> Dict[str, Any]:
397    """
398    Given a configuration dictionary, re-apply the values from the
399    accompanying `_symlinks` dictionary.
400
401    Parameters
402    ----------
403    config: Dict[str, Any]
404        The configuration dictionary containing a `_symlinks` dictionary.
405
406    Returns
407    -------
408    A configuration dictionary with `_symlinks` re-applied.
409    """
410    import copy
411    from meerschaum._internal.static import STATIC_CONFIG
412
413    symlinks_key = STATIC_CONFIG['config']['symlinks_key']
414    if symlinks_key not in config:
415        return config
416
417    reverted_config = copy.deepcopy(config)
418    symlinks_config = reverted_config.pop(symlinks_key)
419
420    def deep_patch(target_dict, patch_dict):
421        for key, value in patch_dict.items():
422            if isinstance(value, dict) and key in target_dict and isinstance(target_dict[key], dict):
423                deep_patch(target_dict[key], value)
424            else:
425                target_dict[key] = value
426
427    deep_patch(reverted_config, symlinks_config)
428    return reverted_config