meerschaum.config

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

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

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

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

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

217def search_and_substitute_config(
218    config: Dict[str, Any],
219    leading_key: str = "MRSM",
220    delimiter: str = ":",
221    begin_key: str = "{",
222    end_key: str = "}",
223    literal_key: str = '!',
224    keep_symlinks: bool = True,
225) -> Dict[str, Any]:
226    """
227    Search the config for Meerschaum substitution syntax and substite with value of keys.
228
229    Parameters
230    ----------
231    config: Dict[str, Any]
232        The Meerschaum configuration dictionary to search through.
233
234    leading_key: str, default 'MRSM'
235        The string with which to start the search.
236
237    begin_key: str, default '{'
238        The string to start the keys list.
239
240    end_key: str, default '}'
241        The string to end the keys list.
242
243    literal_key: str, default '!'
244        The string to force an literal interpretation of a value.
245        When the string is isolated, a literal interpreation is assumed and the surrounding
246        quotes are replaced.
247        
248        E.g. Suppose a:b:c produces a dictionary {'d': 1}.
249        - 'MRSM{a:b:c}'    => {'d': 1}        : isolated
250        - ' MRSM{a:b:c} '  => ' "{\'d\': 1}"' : not isolated
251        - ' MRSM{!a:b:c} ' => ' {"d": 1}'     : literal
252
253    keep_symlinks: bool, default True
254        If True, include the symlinks under the top-level key '_symlinks' (never written to a file).
255        Defaults to True.
256        
257        Example:
258
259        ```
260        MRSM{meerschaum:connectors:main:host} => cf['meerschaum']['connectors']['main']['host']
261        ``` 
262
263    Returns
264    -------
265    The configuration dictionary with `MRSM{}` symlinks replaced with
266    the values from the current configuration.
267    """
268    from meerschaum.config import get_config
269
270    _links = []
271    def _find_symlinks(d, _keys: Optional[List[str]] = None):
272        if _keys is None:
273            _keys = []
274        if not isinstance(d, dict):
275            return
276        for k, v in d.items():
277            if isinstance(v, dict):
278                _find_symlinks(v, _keys + [k])
279            elif (leading_key + begin_key) in str(v):
280                _links.append((_keys + [k], v))
281
282    _find_symlinks(config)
283
284    import json
285    needle = leading_key + begin_key
286    haystack = json.dumps(config, separators=(',', ':'))
287
288    patterns = {}
289    isolated_patterns = {}
290    literal_patterns = {}
291
292    begin, end, floor = 0, 0, 0
293    while needle in haystack[floor:]:
294        ### extract the keys
295        hs = haystack[floor:]
296
297        ### the first character of the keys
298        ### MRSM{key1:key2}
299        ###      ^
300        begin = hs.find(needle) + len(needle)
301
302        ### The character behind the needle.
303        ### "MRSM{key1:key2}"
304        ### ^
305        prior = haystack[(floor + begin) - (len(needle) + 1)]
306
307        ### number of characters to end of keys
308        ### (really it's the index of the beginning of the end_key relative to the beginning
309        ###     but the math works out)
310        ### MRSM{key1}
311        ###      ^   ^  => 4
312        length = hs[begin:].find(end_key)
313
314        ### index of the end_key (end of `length` characters)
315        end = begin + length
316
317        ### The character after the end_key.
318        after = haystack[floor + end + 1]
319
320        ### advance the floor to find the next leading key
321        floor += end + len(end_key)
322        pattern_keys = hs[begin:end].split(delimiter)
323
324        ### Check for isolation key and empty keys (MRSM{}).
325        force_literal = False
326        keys = [k for k in pattern_keys]
327        if str(keys[0]).startswith(literal_key):
328            keys[0] = str(keys[0])[len(literal_key):]
329            force_literal = True
330        if len(keys) == 1 and keys[0] == '':
331            keys = []
332
333        ### Evaluate the parsed keys to extract the referenced value.
334        ### TODO This needs to be recursive for chaining symlinks together.
335        try:
336            valid, value = get_config(
337                *keys,
338                substitute=False,
339                as_tuple=True,
340                write_missing=False,
341                sync_files=False,
342            )
343        except Exception:
344            import traceback
345            traceback.print_exc()
346            valid = False
347        if not valid:
348            continue
349
350        ### pattern to search and replace
351        pattern = leading_key + begin_key + delimiter.join(pattern_keys) + end_key
352
353        ### store patterns and values
354        patterns[pattern] = value
355
356        ### Determine whether the pattern occured inside a string or is an isolated, direct symlink.
357        isolated_patterns[pattern] = (prior == '"' and after == '"')
358
359        literal_patterns[pattern] = force_literal
360
361    ### replace the patterns with the values
362    for pattern, value in patterns.items():
363        if isolated_patterns[pattern]:
364            haystack = haystack.replace(
365                json.dumps(pattern),
366                json.dumps(value),
367            )
368        elif literal_patterns[pattern]:
369            haystack = haystack.replace(
370                pattern,
371                (
372                    json.dumps(value)
373                    .replace("\\", "\\\\")
374                    .replace('"', '\\"')
375                    .replace("'", "\\'")
376                )
377            )
378        else:
379            haystack = haystack.replace(pattern, str(value))
380
381    ### parse back into dict
382    parsed_config = json.loads(haystack) or {}
383
384    symlinks = {}
385    if keep_symlinks:
386        ### Keep track of symlinks for writing back to a file.
387        for _keys, _pattern in _links:
388            s = symlinks
389            for k in _keys[:-1]:
390                if k not in s:
391                    s[k] = {}
392                s = s[k]
393            s[_keys[-1]] = _pattern
394
395        from meerschaum.config._patch import apply_patch_to_config
396        symlinks_key = STATIC_CONFIG['config']['symlinks_key']
397        if symlinks:
398            if symlinks_key not in parsed_config:
399                parsed_config[symlinks_key] = symlinks
400            else:
401                parsed_config[symlinks_key] = apply_patch_to_config(
402                    parsed_config[symlinks_key],
403                    symlinks,
404                    warn=False,
405                )
406
407    return parsed_config

410def revert_symlinks_config(config: Dict[str, Any]) -> Dict[str, Any]:
411    """
412    Given a configuration dictionary, re-apply the values from the
413    accompanying `_symlinks` dictionary.
414
415    Parameters
416    ----------
417    config: Dict[str, Any]
418        The configuration dictionary containing a `_symlinks` dictionary.
419
420    Returns
421    -------
422    A configuration dictionary with `_symlinks` re-applied.
423    """
424    import copy
425    from meerschaum._internal.static import STATIC_CONFIG
426
427    symlinks_key = STATIC_CONFIG['config']['symlinks_key']
428    if symlinks_key not in config:
429        return config
430
431    reverted_config = copy.deepcopy(config)
432    symlinks_config = reverted_config.pop(symlinks_key)
433
434    def deep_patch(target_dict, patch_dict):
435        for key, value in patch_dict.items():
436            if isinstance(value, dict) and key in target_dict and isinstance(target_dict[key], dict):
437                deep_patch(target_dict[key], value)
438            else:
439                target_dict[key] = value
440
441    deep_patch(reverted_config, symlinks_config)
442    return reverted_config

445def get_possible_keys() -> List[str]:
446    """
447    Return a list of possible top-level keys.
448    """
449    import os
450    from meerschaum.config._paths import CONFIG_DIR_PATH
451    from meerschaum.config._default import default_config
452    keys = set()
453    for key in default_config:
454        keys.add(key)
455    for filename in os.listdir(CONFIG_DIR_PATH):
456        keys.add('.'.join(filename.split('.')[:-1]))
457    return sorted(list(keys))

460def get_keyfile_path(
461    key: str,
462    create_new: bool = False,
463    directory: Union[pathlib.Path, str, None] = None,
464) -> Union[pathlib.Path, None]:
465    """Determine a key's file path."""
466    import os
467    import pathlib
468    if directory is None:
469        from meerschaum.config._paths import CONFIG_DIR_PATH
470        directory = CONFIG_DIR_PATH
471
472    try:
473        return pathlib.Path(
474            os.path.join(
475                directory,
476                read_config(
477                    keys=[key],
478                    with_filenames=True,
479                    write_missing=False,
480                    substitute=False,
481                )[1][0]
482            )
483        )
484    except IndexError:
485        if create_new:
486            default_filetype = STATIC_CONFIG['config']['default_filetype']
487            return pathlib.Path(os.path.join(directory, key + '.' + default_filetype))
488        return None