meerschaum.config

173def get_plugin_config(
174    *keys: str,
175    warn: bool = False,
176    **kw: Any
177) -> Optional[Any]:
178    """
179    This may only be called from within a Meerschaum plugin.
180    See `meerschaum.config.get_config` for arguments.
181    """
182    from meerschaum.utils.warnings import error
183    from meerschaum.plugins import _get_parent_plugin
184    parent_plugin_name = _get_parent_plugin(2)
185    if parent_plugin_name is None:
186        error(
187            "You may only call `get_plugin_config()` "
188            "from within a Meerschaum plugin."
189        )
190
191    return _core.get_plugin_config(parent_plugin_name, *keys, warn=warn, **kw)

194def write_plugin_config(
195    config_dict: Dict[str, Any],
196    **kw: Any
197):
198    """
199    Write a plugin's configuration dictionary.
200    """
201    from meerschaum.utils.warnings import error
202    from meerschaum.plugins import _get_parent_plugin
203    parent_plugin_name = _get_parent_plugin(2)
204    if parent_plugin_name is None:
205        error("You may only call `get_plugin_config()` from within a Meerschaum plugin.")
206    return _core.write_plugin_config(parent_plugin_name, config_dict, **kw)

116def get_config(
117    *keys: str,
118    patch: bool = True,
119    substitute: bool = True,
120    sync_files: bool = True,
121    write_missing: bool = True,
122    as_tuple: bool = False,
123    warn: bool = True,
124    debug: bool = False
125) -> Any:
126    """
127    Return the Meerschaum configuration dictionary.
128    If positional arguments are provided, index by the keys.
129    Raises a warning if invalid keys are provided.
130
131    Parameters
132    ----------
133    keys: str:
134        List of strings to index.
135
136    patch: bool, default True
137        If `True`, patch missing default keys into the config directory.
138        Defaults to `True`.
139
140    sync_files: bool, default True
141        If `True`, sync files if needed.
142        Defaults to `True`.
143
144    write_missing: bool, default True
145        If `True`, write default values when the main config files are missing.
146        Defaults to `True`.
147
148    substitute: bool, default True
149        If `True`, subsitute 'MRSM{}' values.
150        Defaults to `True`.
151
152    as_tuple: bool, default False
153        If `True`, return a tuple of type (success, value).
154        Defaults to `False`.
155        
156    Returns
157    -------
158    The value in the configuration directory, indexed by the provided keys.
159
160    Examples
161    --------
162    >>> get_config('meerschaum', 'instance')
163    'sql:main'
164    >>> get_config('does', 'not', 'exist')
165    UserWarning: Invalid keys in config: ('does', 'not', 'exist')
166    """
167    return _core.get_config(
168        *keys, patch=patch, substitute=substitute, sync_files=sync_files,
169        write_missing=write_missing, as_tuple=as_tuple, warn=warn, debug=debug,
170    )

101def write_config(
102    config_dict: Optional[Dict[str, Any]] = None,
103    directory: Union[str, pathlib.Path, None] = None,
104    debug: bool = False,
105    **kw: Any
106) -> bool:
107    """Write YAML and JSON files to the configuration directory.
108
109    Parameters
110    ----------
111    config_dict: Optional[Dict[str, Any]], default None
112        A dictionary of keys to dictionaries of configuration.
113        Each key corresponds to a .yaml or .json config file.
114        Writing config to a directory with different keys
115        does not affect existing keys in that directory.
116        If not provided, use the currently loaded config dictionary.
117
118    directory: Union[str, pathlib.Path, None], default None
119        The directory to which the keys are written.
120        If not provided, use the default config path (`~/.config/meerschaum/config/`).
121
122    Returns
123    -------
124    A bool indicating success.
125
126    """
127    _rust = _core.write_config
128    return _rust(config_dict, directory=directory, debug=debug)

17def edit_config(
18    keys: Optional[List[str]] = None,
19    params: Optional[Dict[str, Any]] = None,
20    debug: bool = False,
21    **kw: Any
22) -> SuccessTuple:
23    """Edit the configuration files."""
24    import meerschaum.config.paths as paths
25    from meerschaum.config import get_config, _config
26    from meerschaum.config._read_config import get_keyfile_path, read_config, revert_symlinks_config
27    from meerschaum.config._backup import backup_config_key, restore_config_backup
28    from meerschaum._internal.static import STATIC_CONFIG
29    from meerschaum.utils.packages import reload_meerschaum
30    from meerschaum.utils.misc import edit_file
31    from meerschaum.utils.warnings import warn, info
32    from meerschaum.utils.prompt import prompt, yes_no
33
34    if keys is None:
35        keys = []
36
37    symlinks_key = STATIC_CONFIG['config']['symlinks_key']
38    def _edit_key(key: str):
39        new_key_config = None
40        ### Snapshot the pristine file before any edits or default-merging so it
41        ### can be rolled back if the edit corrupts it.
42        backup_path = backup_config_key(key, debug=debug)
43        while True:
44            ### If defined in default, create the config file.
45            config = _config()
46            symlinks_key_config = config.get(symlinks_key, {}).get(key, {})
47            key_config = revert_symlinks_config({
48                key: config.pop(key, {}),
49                symlinks_key: {key: symlinks_key_config},
50            })
51            keyfile_path = get_keyfile_path(key, create_new=True)
52            get_config(key, write_missing=True, warn=False)
53
54            edit_file(get_keyfile_path(key, create_new=True))
55
56            ### Parse the edited file; offer the backup when it does not parse.
57            try:
58                new_key_config = read_config(
59                    paths.CONFIG_DIR_PATH,
60                    [key],
61                    write_missing=False,
62                    raise_parsing_errors=True,
63                )
64            except Exception:
65                if key_config:
66                    _config()[key] = key_config
67                warn(f"Could not parse key '{key}'.", stack=False)
68                if backup_path is not None and yes_no(
69                    f"Restore '{key}' from the backup taken before editing?",
70                    default='n',
71                ):
72                    restore_success, restore_msg = restore_config_backup(
73                        key, backup_path=backup_path, reload=False, debug=debug,
74                    )
75                    if restore_success:
76                        info(restore_msg)
77                        new_key_config = read_config(
78                            paths.CONFIG_DIR_PATH,
79                            [key],
80                            write_missing=False,
81                            raise_parsing_errors=False,
82                        )
83                        break
84                    warn(restore_msg, stack=False)
85                _ = prompt(f"Press [Enter] to edit '{keyfile_path}', [CTRL+C] to exit.")
86                continue
87
88            if new_key_config:
89                break
90
91    try:
92        for k in keys:
93            _edit_key(k)
94    except KeyboardInterrupt:
95        return False, ""
96
97    reload_meerschaum(debug=debug)
98    return (True, "Success")

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

209@contextlib.contextmanager
210def replace_config(config_: Union[Dict[str, Any], None]):
211    """
212    Temporarily override the Meerschaum config dictionary.
213
214    Parameters
215    ----------
216    config_: Dict[str, Any]
217        The new config dictionary to temporarily replace the canonical `config`.
218    """
219    if config_ is None:
220        yield
221        return
222
223    global _backup_config, _allow_write_missing
224
225    previous_config = _config()
226    previous_backup_config = _backup_config
227    previous_allow_write_missing = _allow_write_missing
228    _backup_config = (
229        previous_backup_config
230        if previous_backup_config is not None
231        else previous_config
232    )
233    _allow_write_missing = False
234    set_config(config_)
235
236    try:
237        yield
238    finally:
239        set_config(previous_config)
240        _backup_config = previous_backup_config
241        _allow_write_missing = previous_allow_write_missing

 65def search_and_substitute_config(
 66    config: Dict[str, Any],
 67    leading_key: str = "MRSM",
 68    delimiter: str = ":",
 69    begin_key: str = "{",
 70    end_key: str = "}",
 71    literal_key: str = '!',
 72    keep_symlinks: bool = True,
 73) -> Dict[str, Any]:
 74    """
 75    Search the config for Meerschaum substitution syntax and substite with value of keys.
 76
 77    Parameters
 78    ----------
 79    config: Dict[str, Any]
 80        The Meerschaum configuration dictionary to search through.
 81
 82    leading_key: str, default 'MRSM'
 83        The string with which to start the search.
 84
 85    begin_key: str, default '{'
 86        The string to start the keys list.
 87
 88    end_key: str, default '}'
 89        The string to end the keys list.
 90
 91    literal_key: str, default '!'
 92        The string to force an literal interpretation of a value.
 93        When the string is isolated, a literal interpreation is assumed and the surrounding
 94        quotes are replaced.
 95        
 96        E.g. Suppose a:b:c produces a dictionary {'d': 1}.
 97        - 'MRSM{a:b:c}'    => {'d': 1}        : isolated
 98        - ' MRSM{a:b:c} '  => ' "{\'d\': 1}"' : not isolated
 99        - ' MRSM{!a:b:c} ' => ' {"d": 1}'     : literal
100
101    keep_symlinks: bool, default True
102        If True, include the symlinks under the top-level key '_symlinks' (never written to a file).
103        Defaults to True.
104        
105        Example:
106
107        ```
108        MRSM{meerschaum:connectors:main:host} => cf['meerschaum']['connectors']['main']['host']
109        ``` 
110
111    Returns
112    -------
113    The configuration dictionary with `MRSM{}` symlinks replaced with
114    the values from the current configuration.
115    """
116    _rust = _core.search_and_substitute_config
117    return _rust(
118        config,
119        leading_key=leading_key,
120        delimiter=delimiter,
121        begin_key=begin_key,
122        end_key=end_key,
123        literal_key=literal_key,
124        keep_symlinks=keep_symlinks,
125    )

128def revert_symlinks_config(config: Dict[str, Any]) -> Dict[str, Any]:
129    """
130    Given a configuration dictionary, re-apply the values from the
131    accompanying `_symlinks` dictionary.
132
133    Parameters
134    ----------
135    config: Dict[str, Any]
136        The configuration dictionary containing a `_symlinks` dictionary.
137
138    Returns
139    -------
140    A configuration dictionary with `_symlinks` re-applied.
141    """
142    _rust = _core.revert_symlinks_config
143    return _rust(config)

146def get_possible_keys() -> List[str]:
147    """
148    Return a list of possible top-level keys.
149    """
150    import os
151    import meerschaum.config.paths as paths
152    from meerschaum.config._default import default_config
153    keys = set()
154    for key in default_config:
155        keys.add(key)
156    for filename in os.listdir(paths.CONFIG_DIR_PATH):
157        keys.add('.'.join(filename.split('.')[:-1]))
158    return sorted(list(keys))

161def get_keyfile_path(
162    key: str,
163    create_new: bool = False,
164    directory: Union[pathlib.Path, str, None] = None,
165) -> Union[pathlib.Path, None]:
166    """Determine a key's file path."""
167    _rust = _core.get_keyfile_path
168    return _rust(key, create_new=create_new, directory=directory)

15def apply_patch_to_config(
16    config: Dict[str, Any],
17    patch: Dict[str, Any],
18    warn: bool = False,
19) -> Dict[str, Any]:
20    """Patch the config dict with a new dict (cascade patching)."""
21    if not isinstance(patch, dict) or not patch:
22        return config
23    if not isinstance(config, dict) or not config:
24        return copy.deepcopy(patch)
25
26    base = config.copy()
27
28    for key, value in patch.items():
29        if isinstance(value, dict) and isinstance(base.get(key), dict):
30            base[key] = apply_patch_to_config(base[key], value, warn=warn)
31        elif isinstance(value, dict):
32            if warn and key in base:
33                _warn(f"Overwriting the value {base[key]} with a dictionary:\n{value}")
34            base[key] = copy.deepcopy(value)
35        else:
36            base[key] = value
37
38    return base

17def read_config(
18    directory: Union[pathlib.Path, str, None] = None,
19    keys: Optional[List[str]] = None,
20    write_missing: bool = True,
21    substitute: bool = True,
22    with_filenames: bool = False,
23    raise_parsing_errors: bool = False,
24) -> Union[Dict[str, Any], Tuple[Dict[str, Any], List[str]]]:
25    """
26    Read the configuration directory.
27
28    Parameters
29    ----------
30    directory: Union[pathlib.Path, str, None], default None
31        The directory with configuration files (.json and .yaml).
32
33    keys: Optional[List[str]], default None
34        Which configuration files to read.
35
36    write_missing: bool, default True
37        If a keyfile does not exist but is defined in the default configuration,
38        write the file to disk.
39
40    substitute: bool, default True
41        Replace `MRSM{}` syntax with configuration values.
42
43    with_filename: bool, default False
44        If `True`, return a tuple of the configuration dictionary with a list of read filenames.
45
46    raise_parsing_errors: bool, default False
47        If `True`, re-raise parsing exceptions.
48
49    Examples
50    --------
51    >>> read_config(keys=['meerschaum'], with_filename=True)
52    >>> ({...}, ['meerschaum.yaml'])
53    """
54    _rust = _core.read_config
55    return _rust(
56        directory=directory,
57        keys=keys,
58        write_missing=write_missing,
59        substitute=substitute,
60        with_filenames=with_filenames,
61        raise_parsing_errors=raise_parsing_errors,
62    )