29def get_pipes(
 30    connector_keys: Union[str, List[str], None] = None,
 31    metric_keys: Union[str, List[str], None] = None,
 32    location_keys: Union[str, List[str], None] = None,
 33    tags: Optional[List[str]] = None,
 34    params: Optional[Dict[str, Any]] = None,
 35    mrsm_instance: Union[str, InstanceConnector, None] = None,
 36    instance: Union[str, InstanceConnector, None] = None,
 37    as_list: bool = False,
 38    as_tags_dict: bool = False,
 39    method: str = 'registered',
 40    workers: Optional[int] = None,
 41    debug: bool = False,
 42    _cache_parameters: bool = True,
 43    **kw: Any
 44) -> Union[PipesDict, List[mrsm.Pipe], Dict[str, mrsm.Pipe]]:
 45    """
 46    Return a dictionary or list of `meerschaum.Pipe` objects.
 47
 48    Parameters
 49    ----------
 50    connector_keys: Union[str, List[str], None], default None
 51        String or list of connector keys.
 52        If omitted or is `'*'`, fetch all possible keys.
 53        If a string begins with `'_'`, select keys that do NOT match the string.
 54
 55    metric_keys: Union[str, List[str], None], default None
 56        String or list of metric keys. See `connector_keys` for formatting.
 57
 58    location_keys: Union[str, List[str], None], default None
 59        String or list of location keys. See `connector_keys` for formatting.
 60
 61    tags: Optional[List[str]], default None
 62         If provided, only include pipes with these tags.
 63
 64    params: Optional[Dict[str, Any]], default None
 65        Dictionary of additional parameters to search by.
 66        Params are parsed into a SQL WHERE clause.
 67        E.g. `{'a': 1, 'b': 2}` equates to `'WHERE a = 1 AND b = 2'`
 68
 69    mrsm_instance: Union[str, InstanceConnector, None], default None
 70        Connector keys for the Meerschaum instance of the pipes.
 71        Must be a `meerschaum.connectors.sql.SQLConnector.SQLConnector` or
 72        `meerschaum.connectors.api.APIConnector.APIConnector`.
 73        
 74    as_list: bool, default False
 75        If `True`, return pipes in a list instead of a hierarchical dictionary.
 76        `False` : `{connector_keys: {metric_key: {location_key: Pipe}}}`
 77        `True`  : `[Pipe]`
 78
 79    as_tags_dict: bool, default False
 80        If `True`, return a dictionary mapping tags to pipes.
 81        Pipes with multiple tags will be repeated.
 82
 83    method: str, default 'registered'
 84        Available options: `['registered', 'explicit', 'all']`
 85        If `'registered'` (default), create pipes based on registered keys in the connector's pipes table
 86        (API or SQL connector, depends on mrsm_instance).
 87        If `'explicit'`, create pipes from provided connector_keys, metric_keys, and location_keys
 88        instead of consulting the pipes table. Useful for creating non-existent pipes.
 89        If `'all'`, create pipes from predefined metrics and locations. Required `connector_keys`.
 90        **NOTE:** Method `'all'` is not implemented!
 91
 92    workers: Optional[int], default None
 93        If provided (and `as_tags_dict` is `True`), set the number of workers for the pool
 94        to fetch tags.
 95        Only takes effect if the instance connector supports multi-threading
 96
 97    **kw: Any:
 98        Keyword arguments to pass to the `meerschaum.Pipe` constructor.
 99
100    Returns
101    -------
102    A dictionary of dictionaries and `meerschaum.Pipe` objects
103    in the connector, metric, location hierarchy.
104    If `as_list` is `True`, return a list of `meerschaum.Pipe` objects.
105    If `as_tags_dict` is `True`, return a dictionary mapping tags to pipes.
106
107    Examples
108    --------
109    ```
110    >>> ### Manual definition:
111    >>> pipes = {
112    ...     <connector_keys>: {
113    ...         <metric_key>: {
114    ...             <location_key>: Pipe(
115    ...                 <connector_keys>,
116    ...                 <metric_key>,
117    ...                 <location_key>,
118    ...             ),
119    ...         },
120    ...     },
121    ... },
122    >>> ### Accessing a single pipe:
123    >>> pipes['sql:main']['weather'][None]
124    >>> ### Return a list instead:
125    >>> get_pipes(as_list=True)
126    [Pipe('sql:main', 'weather')]
127    >>> get_pipes(as_tags_dict=True)
128    {'gvl': Pipe('sql:main', 'weather')}
129    ```
130    """
131
132    import json
133    from collections import defaultdict
134    from meerschaum.config import get_config
135    from meerschaum.utils.warnings import error
136    from meerschaum.utils.misc import filter_keywords
137    from meerschaum.utils.pool import get_pool
138
139    if connector_keys is None:
140        connector_keys = []
141    if metric_keys is None:
142        metric_keys = []
143    if location_keys is None:
144        location_keys = []
145    if params is None:
146        params = {}
147    if tags is None:
148        tags = []
149
150    if isinstance(connector_keys, str):
151        connector_keys = [connector_keys]
152    if isinstance(metric_keys, str):
153        metric_keys = [metric_keys]
154    if isinstance(location_keys, str):
155        location_keys = [location_keys]
156
157    ### Get SQL or API connector (keys come from `connector.fetch_pipes_keys()`).
158    if mrsm_instance is None:
159        mrsm_instance = instance
160    if mrsm_instance is None:
161        mrsm_instance = get_config('meerschaum', 'instance', patch=True)
162    if isinstance(mrsm_instance, str):
163        from meerschaum.connectors.parse import parse_instance_keys
164        connector = parse_instance_keys(keys=mrsm_instance, debug=debug)
165    else:
166        from meerschaum.connectors import instance_types
167        valid_connector = False
168        if hasattr(mrsm_instance, 'type'):
169            if mrsm_instance.type in instance_types:
170                valid_connector = True
171        if not valid_connector:
172            error(f"Invalid instance connector: {mrsm_instance}")
173        connector = mrsm_instance
174    if debug:
175        from meerschaum.utils.debug import dprint
176        dprint(f"Using instance connector: {connector}")
177    if not connector:
178        error(f"Could not create connector from keys: '{mrsm_instance}'")
179
180    ### Get a list of tuples for the keys needed to build pipes.
181    result = fetch_pipes_keys(
182        method,
183        connector,
184        connector_keys = connector_keys,
185        metric_keys = metric_keys,
186        location_keys = location_keys,
187        tags = tags,
188        params = params,
189        workers = workers,
190        debug = debug
191    )
192    if result is None:
193        error("Unable to build pipes!")
194
195    ### Populate the `pipes` dictionary with Pipes based on the keys
196    ### obtained from the chosen `method`.
197    from meerschaum import Pipe
198    pipes = {}
199    for keys_tuple in result:
200        ck, mk, lk = keys_tuple[0], keys_tuple[1], keys_tuple[2]
201        pipe_tags_or_parameters = keys_tuple[3] if len(keys_tuple) == 4 else None
202        pipe_parameters = (
203            pipe_tags_or_parameters
204            if isinstance(pipe_tags_or_parameters, (dict, str))
205            else None
206        )
207        if isinstance(pipe_parameters, str):
208            pipe_parameters = json.loads(pipe_parameters)
209        pipe_tags = (
210            pipe_tags_or_parameters
211            if isinstance(pipe_tags_or_parameters, list)
212            else (
213                pipe_tags_or_parameters.get('tags', [])
214                if isinstance(pipe_tags_or_parameters, dict)
215                else None
216            )
217        )
218
219        if ck not in pipes:
220            pipes[ck] = {}
221
222        if mk not in pipes[ck]:
223            pipes[ck][mk] = {}
224
225        pipe = Pipe(
226            ck, mk, lk,
227            mrsm_instance = connector,
228            parameters = pipe_parameters,
229            tags = pipe_tags,
230            debug = debug,
231            **filter_keywords(Pipe, **kw)
232        )
233        pipe.__dict__['_tags'] = pipe_tags
234        pipes[ck][mk][lk] = pipe
235
236    if not as_list and not as_tags_dict:
237        return pipes
238
239    from meerschaum.utils.misc import flatten_pipes_dict
240    pipes_list = flatten_pipes_dict(pipes)
241    if as_list:
242        return pipes_list
243
244    pool = get_pool(workers=(workers if connector.IS_THREAD_SAFE else 1))
245    def gather_pipe_tags(pipe: mrsm.Pipe) -> Tuple[mrsm.Pipe, List[str]]:
246        _tags = pipe.__dict__.get('_tags', None)
247        gathered_tags = _tags if _tags is not None else pipe.tags
248        return pipe, (gathered_tags or [])
249
250    tags_pipes = defaultdict(lambda: [])
251    pipes_tags = dict(pool.map(gather_pipe_tags, pipes_list))
252    for pipe, tags in pipes_tags.items():
253        for tag in (tags or []):
254            tags_pipes[tag].append(pipe)
255
256    return dict(tags_pipes)

 68def get_connector(
 69    type: str = None,
 70    label: str = None,
 71    refresh: bool = False,
 72    debug: bool = False,
 73    _load_plugins: bool = True,
 74    **kw: Any
 75) -> Connector:
 76    """
 77    Return existing connector or create new connection and store for reuse.
 78    
 79    You can create new connectors if enough parameters are provided for the given type and flavor.
 80
 81    Parameters
 82    ----------
 83    type: Optional[str], default None
 84        Connector type (sql, api, etc.).
 85        Defaults to the type of the configured `instance_connector`.
 86
 87    label: Optional[str], default None
 88        Connector label (e.g. main). Defaults to `'main'`.
 89
 90    refresh: bool, default False
 91        Refresh the Connector instance / construct new object. Defaults to `False`.
 92
 93    kw: Any
 94        Other arguments to pass to the Connector constructor.
 95        If the Connector has already been constructed and new arguments are provided,
 96        `refresh` is set to `True` and the old Connector is replaced.
 97
 98    Returns
 99    -------
100    A new Meerschaum connector (e.g. `meerschaum.connectors.api.APIConnector`,
101    `meerschaum.connectors.sql.SQLConnector`).
102    
103    Examples
104    --------
105    The following parameters would create a new
106    `meerschaum.connectors.sql.SQLConnector` that isn't in the configuration file.
107
108    ```
109    >>> conn = get_connector(
110    ...     type = 'sql',
111    ...     label = 'newlabel',
112    ...     flavor = 'sqlite',
113    ...     database = '/file/path/to/database.db'
114    ... )
115    >>>
116    ```
117
118    """
119    from meerschaum.connectors.parse import parse_instance_keys
120    from meerschaum.config import get_config
121    from meerschaum._internal.static import STATIC_CONFIG
122    from meerschaum.utils.warnings import warn
123    global _loaded_plugin_connectors
124    if isinstance(type, str) and not label and ':' in type:
125        type, label = type.split(':', maxsplit=1)
126
127    if _load_plugins:
128        with _locks['_loaded_plugin_connectors']:
129            if not _loaded_plugin_connectors:
130                load_plugin_connectors()
131                _load_builtin_custom_connectors()
132                _loaded_plugin_connectors = True
133
134    if type is None and label is None:
135        default_instance_keys = get_config('meerschaum', 'instance', patch=True)
136        ### recursive call to get_connector
137        return parse_instance_keys(default_instance_keys)
138
139    ### NOTE: the default instance connector may not be main.
140    ### Only fall back to 'main' if the type is provided by the label is omitted.
141    label = label if label is not None else STATIC_CONFIG['connectors']['default_label']
142
143    ### type might actually be a label. Check if so and raise a warning.
144    if type not in connectors:
145        possibilities, poss_msg = [], ""
146        for _type in get_config('meerschaum', 'connectors'):
147            if type in get_config('meerschaum', 'connectors', _type):
148                possibilities.append(f"{_type}:{type}")
149        if len(possibilities) > 0:
150            poss_msg = " Did you mean"
151            for poss in possibilities[:-1]:
152                poss_msg += f" '{poss}',"
153            if poss_msg.endswith(','):
154                poss_msg = poss_msg[:-1]
155            if len(possibilities) > 1:
156                poss_msg += " or"
157            poss_msg += f" '{possibilities[-1]}'?"
158
159        warn(f"Cannot create Connector of type '{type}'." + poss_msg, stack=False)
160        return None
161
162    if 'sql' not in types:
163        from meerschaum.connectors.plugin import PluginConnector
164        from meerschaum.connectors.valkey import ValkeyConnector
165        with _locks['types']:
166            types.update({
167                'api': APIConnector,
168                'sql': SQLConnector,
169                'plugin': PluginConnector,
170                'valkey': ValkeyConnector,
171            })
172
173    ### determine if we need to call the constructor
174    if not refresh:
175        ### see if any user-supplied arguments differ from the existing instance
176        if label in connectors[type]:
177            warning_message = None
178            for attribute, value in kw.items():
179                if attribute not in connectors[type][label].meta:
180                    import inspect
181                    cls = connectors[type][label].__class__
182                    cls_init_signature = inspect.signature(cls)
183                    cls_init_params = cls_init_signature.parameters
184                    if attribute not in cls_init_params:
185                        warning_message = (
186                            f"Received new attribute '{attribute}' not present in connector " +
187                            f"{connectors[type][label]}.\n"
188                        )
189                elif connectors[type][label].__dict__[attribute] != value:
190                    warning_message = (
191                        f"Mismatched values for attribute '{attribute}' in connector "
192                        + f"'{connectors[type][label]}'.\n" +
193                        f"  - Keyword value: '{value}'\n" +
194                        f"  - Existing value: '{connectors[type][label].__dict__[attribute]}'\n"
195                    )
196            if warning_message is not None:
197                warning_message += (
198                    "\nSetting `refresh` to True and recreating connector with type:"
199                    + f" '{type}' and label '{label}'."
200                )
201                refresh = True
202                warn(warning_message)
203        else: ### connector doesn't yet exist
204            refresh = True
205
206    ### only create an object if refresh is True
207    ### (can be manually specified, otherwise determined above)
208    if refresh:
209        with _locks['connectors']:
210            try:
211                ### will raise an error if configuration is incorrect / missing
212                conn = types[type](label=label, **kw)
213                connectors[type][label] = conn
214            except InvalidAttributesError as ie:
215                warn(
216                    f"Incorrect attributes for connector '{type}:{label}'.\n"
217                    + str(ie),
218                    stack = False,
219                )
220                conn = None
221            except Exception as e:
222                from meerschaum.utils.formatting import get_console
223                console = get_console()
224                if console:
225                    console.print_exception()
226                warn(
227                    f"Exception when creating connector '{type}:{label}'.\n" + str(e),
228                    stack = False,
229                )
230                conn = None
231        if conn is None:
232            return None
233
234    return connectors[type][label]

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

 66class Pipe:
 67    """
 68    Access Meerschaum pipes via Pipe objects.
 69    
 70    Pipes are identified by the following:
 71
 72    1. Connector keys (e.g. `'sql:main'`)
 73    2. Metric key (e.g. `'weather'`)
 74    3. Location (optional; e.g. `None`)
 75    
 76    A pipe's connector keys correspond to a data source, and when the pipe is synced,
 77    its `fetch` definition is evaluated and executed to produce new data.
 78    
 79    Alternatively, new data may be directly synced via `pipe.sync()`:
 80    
 81    ```
 82    >>> from meerschaum import Pipe
 83    >>> pipe = Pipe('csv', 'weather')
 84    >>>
 85    >>> import pandas as pd
 86    >>> df = pd.read_csv('weather.csv')
 87    >>> pipe.sync(df)
 88    ```
 89    """
 90
 91    from ._fetch import (
 92        fetch,
 93        get_backtrack_interval,
 94    )
 95    from ._data import (
 96        get_data,
 97        get_backtrack_data,
 98        get_rowcount,
 99        get_data,
100        get_doc,
101        get_value,
102        _get_data_as_iterator,
103        get_chunk_interval,
104        get_chunk_bounds,
105        get_chunk_bounds_batches,
106        parse_date_bounds,
107    )
108    from ._register import register
109    from ._attributes import (
110        attributes,
111        parameters,
112        columns,
113        indices,
114        indexes,
115        dtypes,
116        autoincrement,
117        autotime,
118        upsert,
119        static,
120        tzinfo,
121        enforce,
122        null_indices,
123        mixed_numerics,
124        get_columns,
125        get_columns_types,
126        get_columns_indices,
127        get_indices,
128        get_parameters,
129        get_dtypes,
130        update_parameters,
131        tags,
132        get_id,
133        id,
134        get_val_column,
135        parents,
136        parent,
137        children,
138        child,
139        reference,
140        references,
141        target,
142        _target_legacy,
143        guess_datetime,
144        precision,
145        get_precision,
146    )
147    from ._cache import (
148        _get_cache_connector,
149        _cache_value,
150        _get_cached_value,
151        _invalidate_cache,
152        _get_cache_dir_path,
153        _write_cache_key,
154        _write_cache_file,
155        _write_cache_conn_key,
156        _read_cache_key,
157        _read_cache_file,
158        _read_cache_conn_key,
159        _load_cache_keys,
160        _load_cache_files,
161        _load_cache_conn_keys,
162        _get_cache_keys,
163        _get_cache_file_keys,
164        _get_cache_conn_keys,
165        _clear_cache_key,
166        _clear_cache_file,
167        _clear_cache_conn_key,
168    )
169    from ._show import show
170    from ._edit import edit, edit_definition, update
171    from ._sync import (
172        sync,
173        get_sync_time,
174        exists,
175        filter_existing,
176        _get_chunk_label,
177        get_num_workers,
178        _persist_new_special_columns,
179    )
180    from ._verify import (
181        verify,
182        get_bound_interval,
183        get_bound_time,
184    )
185    from ._delete import delete
186    from ._drop import drop, drop_indices
187    from ._index import create_indices
188    from ._clear import clear
189    from ._deduplicate import deduplicate
190    from ._bootstrap import bootstrap
191    from ._dtypes import enforce_dtypes, infer_dtypes
192    from ._copy import copy_to
193
194    def __init__(
195        self,
196        connector: str = '',
197        metric: str = '',
198        location: Optional[str] = None,
199        parameters: Optional[Dict[str, Any]] = None,
200        columns: Union[Dict[str, str], List[str], None] = None,
201        indices: Optional[Dict[str, Union[str, List[str]]]] = None,
202        tags: Optional[List[str]] = None,
203        target: Optional[str] = None,
204        dtypes: Optional[Dict[str, str]] = None,
205        instance: Optional[Union[str, InstanceConnector]] = None,
206        upsert: Optional[bool] = None,
207        autoincrement: Optional[bool] = None,
208        autotime: Optional[bool] = None,
209        precision: Union[str, Dict[str, Union[str, int]], None] = None,
210        static: Optional[bool] = None,
211        enforce: Optional[bool] = None,
212        null_indices: Optional[bool] = None,
213        mixed_numerics: Optional[bool] = None,
214        temporary: bool = False,
215        cache: Optional[bool] = None,
216        cache_connector_keys: Optional[str] = None,
217        references: Optional[List[Union[str, Dict[str, Any], mrsm.Pipe, None]]] = None,
218        reference: Union[str, Dict[str, Any], mrsm.Pipe, None] = None,
219        parents: Optional[List[Union[str, Dict[str, Any], mrsm.Pipe, None]]] = None,
220        parent: Union[str, Dict[str, Any], mrsm.Pipe, None] = None,
221        children: Optional[List[Union[str, Dict[str, Any], mrsm.Pipe, None]]] = None,
222        child: Union[str, Dict[str, Any], mrsm.Pipe, None] = None,
223        mrsm_instance: Optional[Union[str, InstanceConnector]] = None,
224        connector_keys: Optional[str] = None,
225        metric_key: Optional[str] = None,
226        location_key: Optional[str] = None,
227        instance_keys: Optional[str] = None,
228        indexes: Union[Dict[str, str], List[str], None] = None,
229        debug: bool = False,
230    ):
231        """
232        Parameters
233        ----------
234        connector: str
235            Keys for the pipe's source connector, e.g. `'sql:main'`.
236
237        metric: str
238            Label for the pipe's contents, e.g. `'weather'`.
239
240        location: str, default None
241            Label for the pipe's location. Defaults to `None`.
242
243        parameters: Optional[Dict[str, Any]], default None
244            Optionally set a pipe's parameters from the constructor,
245            e.g. columns and other attributes.
246            You can edit these parameters with `edit pipes`.
247
248        columns: Union[Dict[str, str], List[str], None], default None
249            Set the `columns` dictionary of `parameters`.
250            If `parameters` is also provided, this dictionary is added under the `'columns'` key.
251
252        indices: Optional[Dict[str, Union[str, List[str]]]], default None
253            Set the `indices` dictionary of `parameters`.
254            If `parameters` is also provided, this dictionary is added under the `'indices'` key.
255
256        tags: Optional[List[str]], default None
257            A list of strings to be added under the `'tags'` key of `parameters`.
258            You can select pipes with certain tags using `--tags`.
259
260        dtypes: Optional[Dict[str, str]], default None
261            Set the `dtypes` dictionary of `parameters`.
262            If `parameters` is also provided, this dictionary is added under the `'dtypes'` key.
263
264        mrsm_instance: Optional[Union[str, InstanceConnector]], default None
265            Connector for the Meerschaum instance where the pipe resides.
266            Defaults to the preconfigured default instance (`'sql:main'`).
267
268        instance: Optional[Union[str, InstanceConnector]], default None
269            Alias for `mrsm_instance`. If `mrsm_instance` is supplied, this value is ignored.
270
271        upsert: Optional[bool], default None
272            If `True`, set `upsert` to `True` in the parameters.
273
274        autoincrement: Optional[bool], default None
275            If `True`, set `autoincrement` in the parameters.
276
277        autotime: Optional[bool], default None
278            If `True`, set `autotime` in the parameters.
279
280        precision: Union[str, Dict[str, Union[str, int]], None], default None
281            If provided, set `precision` in the parameters.
282            This may be either a string (the precision unit) or a dictionary of in the form
283            `{'unit': <unit>, 'interval': <interval>}`.
284            Default is determined by the `datetime` column dtype
285            (e.g. `datetime64[us]` is `microsecond` precision).
286
287        static: Optional[bool], default None
288            If `True`, set `static` in the parameters.
289
290        enforce: Optional[bool], default None
291            If `False`, skip data type enforcement.
292            Default behavior is `True`.
293
294        null_indices: Optional[bool], default None
295            Set to `False` if there will be no null values in the index columns.
296            Defaults to `True`.
297
298        mixed_numerics: bool, default None
299            If `True`, integer columns will be converted to `numeric` when floats are synced.
300            Set to `False` to disable this behavior.
301            Defaults to `True`.
302
303        temporary: bool, default False
304            If `True`, prevent instance tables (pipes, users, plugins) from being created.
305
306        cache: Optional[bool], default None
307            If `True`, cache the pipe's metadata to disk (in addition to in-memory caching).
308            If `cache` is not explicitly `True`, it is set to `False` if `temporary` is `True`.
309            Defaults to `True` (from `None`).
310
311        cache_connector_keys: Optional[str], default None
312            If provided, use the keys to a Valkey connector (e.g. `valkey:main`).
313
314        references: Optional[List[Union[str, Dict[str, Any], mrsm.Pipe, None]]], default None
315            If provided, inherit the parameters of the reference Pipe(s).
316            May be equal to a string of the Pipe constructor, a dictionary of constructor keys,
317            a Pipe itself, or a list of any of these values.
318
319        parents: Optional[List[Union[str, Dict[str, Any], mrsm.Pipe, None]]], default None
320            Set references for parent pipes. See `references` for values.
321
322        children: Optional[List[Union[str, Dict[str, Any], mrsm.Pipe, None]]], default None
323            Set references for child pipes. See `references` for values.
324
325        """
326        from meerschaum.utils.warnings import error, warn
327        if (not connector and not connector_keys) or (not metric and not metric_key):
328            error(
329                "Please provide strings for the connector and metric\n    "
330                + "(first two positional arguments)."
331            )
332
333        ### Fall back to legacy `location_key` just in case.
334        if not location:
335            location = location_key
336
337        if not connector:
338            connector = connector_keys
339
340        if not metric:
341            metric = metric_key
342
343        if location in ('[None]', 'None'):
344            location = None
345
346        from meerschaum._internal.static import STATIC_CONFIG
347        negation_prefix = STATIC_CONFIG['system']['fetch_pipes_keys']['negation_prefix']
348        for k in (connector, metric, location, *(tags or [])):
349            if str(k).startswith(negation_prefix):
350                error(f"A pipe's keys and tags cannot start with the prefix '{negation_prefix}'.")
351
352        self._connector_keys = str(connector)
353        self._connector_key = self.connector_keys ### Alias
354        self._metric_key = metric
355        self._location_key = location
356        self.temporary = temporary
357        self.cache = (
358            cache
359            if cache is not None
360            else ((not temporary) and get_config('pipes', 'cache', 'enabled', warn=False))
361        )
362        self.cache_connector_keys = (
363            str(cache_connector_keys)
364            if cache_connector_keys is not None
365            else None
366        )
367        self.debug = debug
368
369        self._attributes: Dict[str, Any] = {
370            'connector_keys': self._connector_keys,
371            'metric_key': self._metric_key,
372            'location_key': self._location_key,
373            'parameters': {},
374        }
375
376        ### only set parameters if values are provided
377        if isinstance(parameters, dict):
378            self._attributes['parameters'] = parameters
379        else:
380            if parameters is not None:
381                warn(f"The provided parameters are of invalid type '{type(parameters)}'.")
382            self._attributes['parameters'] = {}
383
384        columns = columns or self._attributes.get('parameters', {}).get('columns', None)
385        if isinstance(columns, (list, tuple)):
386            columns = {str(col): str(col) for col in columns}
387        if isinstance(columns, dict):
388            self._attributes['parameters']['columns'] = columns
389        elif isinstance(columns, str) and 'Pipe(' in columns:
390            pass
391        elif columns is not None:
392            warn(f"The provided columns are of invalid type '{type(columns)}'.")
393
394        indices = (
395            indices
396            or indexes
397            or self._attributes.get('parameters', {}).get('indices', None)
398            or self._attributes.get('parameters', {}).get('indexes', None)
399        )
400        if isinstance(indices, dict):
401            indices_key = (
402                'indexes'
403                if 'indexes' in self._attributes['parameters']
404                else 'indices'
405            )
406            self._attributes['parameters'][indices_key] = indices
407
408        if isinstance(tags, (list, tuple)):
409            self._attributes['parameters']['tags'] = tags
410        elif tags is not None:
411            warn(f"The provided tags are of invalid type '{type(tags)}'.")
412
413        if isinstance(target, str):
414            self._attributes['parameters']['target'] = target
415        elif target is not None:
416            warn(f"The provided target is of invalid type '{type(target)}'.")
417
418        if isinstance(dtypes, dict):
419            self._attributes['parameters']['dtypes'] = dtypes
420        elif dtypes is not None:
421            warn(f"The provided dtypes are of invalid type '{type(dtypes)}'.")
422
423        if isinstance(upsert, bool):
424            self._attributes['parameters']['upsert'] = upsert
425
426        if isinstance(autoincrement, bool):
427            self._attributes['parameters']['autoincrement'] = autoincrement
428
429        if isinstance(autotime, bool):
430            self._attributes['parameters']['autotime'] = autotime
431
432        if isinstance(precision, dict):
433            self._attributes['parameters']['precision'] = precision
434        elif isinstance(precision, str):
435            self._attributes['parameters']['precision'] = {'unit': precision}
436
437        if isinstance(static, bool):
438            self._attributes['parameters']['static'] = static
439            self._static = static
440
441        if isinstance(enforce, bool):
442            self._attributes['parameters']['enforce'] = enforce
443
444        if isinstance(null_indices, bool):
445            self._attributes['parameters']['null_indices'] = null_indices
446
447        if isinstance(mixed_numerics, bool):
448            self._attributes['parameters']['mixed_numerics'] = mixed_numerics
449
450        ### NOTE: The parameters dictionary is {} by default.
451        ###       A Pipe may be registered without parameters, then edited,
452        ###       or a Pipe may be registered with parameters set in-memory first.
453        _mrsm_instance = mrsm_instance if mrsm_instance is not None else (instance or instance_keys)
454        if _mrsm_instance is None:
455            _mrsm_instance = get_config('meerschaum', 'instance', patch=True)
456
457        if not isinstance(_mrsm_instance, str):
458            self._instance_connector = _mrsm_instance
459            self._instance_keys = str(_mrsm_instance)
460        else:
461            self._instance_keys = _mrsm_instance
462
463        if self._instance_keys == 'sql:memory':
464            self.cache = False
465
466        self._cache_locks = collections.defaultdict(lambda: threading.RLock())
467
468        if references is not None or reference is not None:
469            reference_vals = references if references is not None else reference
470            self.references = reference_vals
471
472        if parents is not None or parent is not None:
473            parent_vals = parents if parents is not None else parent
474            self.parents = parent_vals
475
476        if children is not None or child is not None:
477            children_vals = children if children is not None else child
478            self.children = children_vals
479
480    @property
481    def metric_key(self) -> str:
482        """
483        Return the pipe's metric key.
484        """
485        return self._metric_key
486
487    @property
488    def metric(self) -> str:
489        """
490        Return the pipe's metric key.
491        """
492        return self._metric_key
493
494    @property
495    def location_key(self) -> Union[str, None]:
496        """
497        Return the pipe's location key.
498        """
499        return self._location_key
500
501    @property
502    def location(self) -> Union[str, None]:
503        """
504        Return the pipe's location key.
505        """
506        return self._location_key
507
508    @property
509    def meta(self):
510        """
511        Return the four keys needed to reconstruct this pipe.
512        """
513        return {
514            'connector_keys': self.connector_keys,
515            'metric_key': self.metric_key,
516            'location_key': self.location_key,
517            'instance_keys': self.instance_keys,
518        }
519
520    def keys(self) -> List[str]:
521        """
522        Return the ordered keys for this pipe.
523        """
524        return {
525            key: val
526            for key, val in self.meta.items()
527            if key != 'instance'
528        }
529
530    @property
531    def instance_keys(self) -> str:
532        """
533        Return the pipe's instance keys.
534        """
535        return self._instance_keys
536
537    @property
538    def instance(self) -> Union[InstanceConnector, str]:
539        """
540        Return the pipe's instance connector or keys.
541        """
542        conn = self.instance_connector
543        if conn is None:
544            return self.instance_keys
545        return conn
546
547    @property
548    def instance_connector(self) -> Union[InstanceConnector, None]:
549        """
550        The instance connector on which this pipe resides.
551        """
552        if '_instance_connector' not in self.__dict__:
553            from meerschaum.connectors.parse import parse_instance_keys
554            conn = parse_instance_keys(self.instance_keys)
555            if conn:
556                self._instance_connector = conn
557            else:
558                return None
559        return self._instance_connector
560
561    @property
562    def connector_keys(self) -> str:
563        """
564        Return the pipe's connector keys.
565        """
566        return self._connector_keys
567
568    @property
569    def connector_key(self) -> str:
570        """
571        Legacy: use `Pipe.connector_keys` instead.
572        """
573        return self.connector_keys
574
575    @property
576    def connector(self) -> Union['Connector', str]:
577        """
578        The connector to the data source.
579        """
580        if '_connector' not in self.__dict__:
581            from meerschaum.connectors.parse import parse_instance_keys
582            import warnings
583            with warnings.catch_warnings():
584                warnings.simplefilter('ignore')
585                try:
586                    conn = parse_instance_keys(self.connector_keys)
587                except Exception:
588                    conn = None
589            if conn:
590                self._connector = conn
591            else:
592                return self._connector_keys
593        return self._connector
594
595    def __str__(self, ansi: bool=False):
596        return pipe_repr(self, ansi=ansi)
597
598    def __eq__(self, other):
599        try:
600            return (
601                isinstance(self, type(other))
602                and self.connector_keys == other.connector_keys
603                and self.metric_key == other.metric_key
604                and self.location_key == other.location_key
605                and self.instance_keys == other.instance_keys
606            )
607        except Exception:
608            return False
609
610    def __hash__(self):
611        ### Using an esoteric separator to avoid collisions.
612        sep = "[\"']"
613        return hash(
614            str(self.connector_keys) + sep
615            + str(self.metric_key) + sep
616            + str(self.location_key) + sep
617            + str(self.instance_keys) + sep
618        )
619
620    def __repr__(self, ansi: bool=True, **kw) -> str:
621        if not hasattr(sys, 'ps1'):
622            ansi = False
623
624        return pipe_repr(self, ansi=ansi, **kw)
625
626    def __pt_repr__(self):
627        from meerschaum.utils.packages import attempt_import
628        prompt_toolkit_formatted_text = attempt_import('prompt_toolkit.formatted_text', lazy=False)
629        return prompt_toolkit_formatted_text.ANSI(pipe_repr(self, ansi=True))
630
631    def __getstate__(self) -> Dict[str, Any]:
632        """
633        Define the state dictionary (pickling).
634        """
635        return {
636            'connector_keys': self.connector_keys,
637            'metric_key': self.metric_key,
638            'location_key': self.location_key,
639            'parameters': self._attributes.get('parameters', None),
640            'instance_keys': self.instance_keys,
641        }
642
643    def __setstate__(self, _state: Dict[str, Any]):
644        """
645        Read the state (unpickling).
646        """
647        self.__init__(**_state)
648
649    def __getitem__(self, key: str) -> Any:
650        """
651        Index the pipe's attributes.
652        If the `key` cannot be found`, return `None`.
653        """
654        if key in self.attributes:
655            return self.attributes.get(key, None)
656
657        aliases = {
658            'connector': 'connector_keys',
659            'connector_key': 'connector_keys',
660            'metric': 'metric_key',
661            'location': 'location_key',
662        }
663        aliased_key = aliases.get(key, None)
664        if aliased_key is not None:
665            return self.attributes.get(aliased_key, None)
666
667        property_aliases = {
668            'instance': 'instance_keys',
669            'instance_key': 'instance_keys',
670        }
671        aliased_key = property_aliases.get(key, None)
672        if aliased_key is not None:
673            key = aliased_key
674        return getattr(self, key, None)
675
676    def __copy__(self):
677        """
678        Return a shallow copy of the current pipe.
679        """
680        return mrsm.Pipe(
681            self.connector_keys, self.metric_key, self.location_key,
682            instance=self.instance_keys,
683            parameters=self._attributes.get('parameters', None),
684        )
685
686    def __deepcopy__(self, memo):
687        """
688        Return a deep copy of the current pipe.
689        """
690        return self.__copy__()