19def get_pipes(
 20    connector_keys: Union[str, List[str], None] = None,
 21    metric_keys: Union[str, List[str], None] = None,
 22    location_keys: Union[str, List[str], None] = None,
 23    tags: Optional[List[str]] = None,
 24    params: Optional[Dict[str, Any]] = None,
 25    mrsm_instance: Union[str, InstanceConnector, None] = None,
 26    instance: Union[str, InstanceConnector, None] = None,
 27    as_list: bool = False,
 28    method: str = 'registered',
 29    debug: bool = False,
 30    **kw: Any
 31) -> Union[PipesDict, List[mrsm.Pipe]]:
 32    """
 33    Return a dictionary or list of `meerschaum.Pipe` objects.
 34
 35    Parameters
 36    ----------
 37    connector_keys: Union[str, List[str], None], default None
 38        String or list of connector keys.
 39        If omitted or is `'*'`, fetch all possible keys.
 40        If a string begins with `'_'`, select keys that do NOT match the string.
 41
 42    metric_keys: Union[str, List[str], None], default None
 43        String or list of metric keys. See `connector_keys` for formatting.
 44
 45    location_keys: Union[str, List[str], None], default None
 46        String or list of location keys. See `connector_keys` for formatting.
 47
 48    tags: Optional[List[str]], default None
 49         If provided, only include pipes with these tags.
 50
 51    params: Optional[Dict[str, Any]], default None
 52        Dictionary of additional parameters to search by.
 53        Params are parsed into a SQL WHERE clause.
 54        E.g. `{'a': 1, 'b': 2}` equates to `'WHERE a = 1 AND b = 2'`
 55
 56    mrsm_instance: Union[str, InstanceConnector, None], default None
 57        Connector keys for the Meerschaum instance of the pipes.
 58        Must be a `meerschaum.connectors.sql.SQLConnector.SQLConnector` or
 59        `meerschaum.connectors.api.APIConnector.APIConnector`.
 60        
 61    as_list: bool, default False
 62        If `True`, return pipes in a list instead of a hierarchical dictionary.
 63        `False` : `{connector_keys: {metric_key: {location_key: Pipe}}}`
 64        `True`  : `[Pipe]`
 65
 66    method: str, default 'registered'
 67        Available options: `['registered', 'explicit', 'all']`
 68        If `'registered'` (default), create pipes based on registered keys in the connector's pipes table
 69        (API or SQL connector, depends on mrsm_instance).
 70        If `'explicit'`, create pipes from provided connector_keys, metric_keys, and location_keys
 71        instead of consulting the pipes table. Useful for creating non-existent pipes.
 72        If `'all'`, create pipes from predefined metrics and locations. Required `connector_keys`.
 73        **NOTE:** Method `'all'` is not implemented!
 74
 75    **kw: Any:
 76        Keyword arguments to pass to the `meerschaum.Pipe` constructor.
 77        
 78
 79    Returns
 80    -------
 81    A dictionary of dictionaries and `meerschaum.Pipe` objects
 82    in the connector, metric, location hierarchy.
 83    If `as_list` is `True`, return a list of `meerschaum.Pipe` objects.
 84
 85    Examples
 86    --------
 87    ```
 88    >>> ### Manual definition:
 89    >>> pipes = {
 90    ...     <connector_keys>: {
 91    ...         <metric_key>: {
 92    ...             <location_key>: Pipe(
 93    ...                 <connector_keys>,
 94    ...                 <metric_key>,
 95    ...                 <location_key>,
 96    ...             ),
 97    ...         },
 98    ...     },
 99    ... },
100    >>> ### Accessing a single pipe:
101    >>> pipes['sql:main']['weather'][None]
102    >>> ### Return a list instead:
103    >>> get_pipes(as_list=True)
104    [sql_main_weather]
105    >>> 
106    ```
107    """
108
109    from meerschaum.config import get_config
110    from meerschaum.utils.warnings import error
111    from meerschaum.utils.misc import filter_keywords
112
113    if connector_keys is None:
114        connector_keys = []
115    if metric_keys is None:
116        metric_keys = []
117    if location_keys is None:
118        location_keys = []
119    if params is None:
120        params = {}
121    if tags is None:
122        tags = []
123
124    if isinstance(connector_keys, str):
125        connector_keys = [connector_keys]
126    if isinstance(metric_keys, str):
127        metric_keys = [metric_keys]
128    if isinstance(location_keys, str):
129        location_keys = [location_keys]
130
131    ### Get SQL or API connector (keys come from `connector.fetch_pipes_keys()`).
132    if mrsm_instance is None:
133        mrsm_instance = instance
134    if mrsm_instance is None:
135        mrsm_instance = get_config('meerschaum', 'instance', patch=True)
136    if isinstance(mrsm_instance, str):
137        from meerschaum.connectors.parse import parse_instance_keys
138        connector = parse_instance_keys(keys=mrsm_instance, debug=debug)
139    else:
140        from meerschaum.connectors import instance_types
141        valid_connector = False
142        if hasattr(mrsm_instance, 'type'):
143            if mrsm_instance.type in instance_types:
144                valid_connector = True
145        if not valid_connector:
146            error(f"Invalid instance connector: {mrsm_instance}")
147        connector = mrsm_instance
148    if debug:
149        from meerschaum.utils.debug import dprint
150        dprint(f"Using instance connector: {connector}")
151    if not connector:
152        error(f"Could not create connector from keys: '{mrsm_instance}'")
153
154    ### Get a list of tuples for the keys needed to build pipes.
155    result = fetch_pipes_keys(
156        method,
157        connector,
158        connector_keys = connector_keys,
159        metric_keys = metric_keys,
160        location_keys = location_keys,
161        tags = tags,
162        params = params,
163        debug = debug
164    )
165    if result is None:
166        error(f"Unable to build pipes!")
167
168    ### Populate the `pipes` dictionary with Pipes based on the keys
169    ### obtained from the chosen `method`.
170    from meerschaum import Pipe
171    pipes = {}
172    for ck, mk, lk in result:
173        if ck not in pipes:
174            pipes[ck] = {}
175
176        if mk not in pipes[ck]:
177            pipes[ck][mk] = {}
178
179        pipes[ck][mk][lk] = Pipe(
180            ck, mk, lk,
181            mrsm_instance = connector,
182            debug = debug,
183            **filter_keywords(Pipe, **kw)
184        )
185
186    if not as_list:
187        return pipes
188    from meerschaum.utils.misc import flatten_pipes_dict
189    return flatten_pipes_dict(pipes)

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

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

 60class Pipe:
 61    """
 62    Access Meerschaum pipes via Pipe objects.
 63    
 64    Pipes are identified by the following:
 65
 66    1. Connector keys (e.g. `'sql:main'`)
 67    2. Metric key (e.g. `'weather'`)
 68    3. Location (optional; e.g. `None`)
 69    
 70    A pipe's connector keys correspond to a data source, and when the pipe is synced,
 71    its `fetch` definition is evaluated and executed to produce new data.
 72    
 73    Alternatively, new data may be directly synced via `pipe.sync()`:
 74    
 75    ```
 76    >>> from meerschaum import Pipe
 77    >>> pipe = Pipe('csv', 'weather')
 78    >>>
 79    >>> import pandas as pd
 80    >>> df = pd.read_csv('weather.csv')
 81    >>> pipe.sync(df)
 82    ```
 83    """
 84
 85    from ._fetch import (
 86        fetch,
 87        get_backtrack_interval,
 88    )
 89    from ._data import (
 90        get_data,
 91        get_backtrack_data,
 92        get_rowcount,
 93        _get_data_as_iterator,
 94        get_chunk_interval,
 95        get_chunk_bounds,
 96        parse_date_bounds,
 97    )
 98    from ._register import register
 99    from ._attributes import (
100        attributes,
101        parameters,
102        columns,
103        indices,
104        indexes,
105        dtypes,
106        autoincrement,
107        upsert,
108        static,
109        tzinfo,
110        enforce,
111        null_indices,
112        get_columns,
113        get_columns_types,
114        get_columns_indices,
115        get_indices,
116        tags,
117        get_id,
118        id,
119        get_val_column,
120        parents,
121        children,
122        target,
123        _target_legacy,
124        guess_datetime,
125    )
126    from ._show import show
127    from ._edit import edit, edit_definition, update
128    from ._sync import (
129        sync,
130        get_sync_time,
131        exists,
132        filter_existing,
133        _get_chunk_label,
134        get_num_workers,
135        _persist_new_json_columns,
136        _persist_new_numeric_columns,
137        _persist_new_uuid_columns,
138        _persist_new_bytes_columns,
139    )
140    from ._verify import (
141        verify,
142        get_bound_interval,
143        get_bound_time,
144    )
145    from ._delete import delete
146    from ._drop import drop, drop_indices
147    from ._index import create_indices
148    from ._clear import clear
149    from ._deduplicate import deduplicate
150    from ._bootstrap import bootstrap
151    from ._dtypes import enforce_dtypes, infer_dtypes
152    from ._copy import copy_to
153
154    def __init__(
155        self,
156        connector: str = '',
157        metric: str = '',
158        location: Optional[str] = None,
159        parameters: Optional[Dict[str, Any]] = None,
160        columns: Union[Dict[str, str], List[str], None] = None,
161        indices: Optional[Dict[str, Union[str, List[str]]]] = None,
162        tags: Optional[List[str]] = None,
163        target: Optional[str] = None,
164        dtypes: Optional[Dict[str, str]] = None,
165        instance: Optional[Union[str, InstanceConnector]] = None,
166        temporary: bool = False,
167        upsert: Optional[bool] = None,
168        autoincrement: Optional[bool] = None,
169        static: Optional[bool] = None,
170        enforce: Optional[bool] = None,
171        null_indices: Optional[bool] = None,
172        mrsm_instance: Optional[Union[str, InstanceConnector]] = None,
173        cache: bool = False,
174        debug: bool = False,
175        connector_keys: Optional[str] = None,
176        metric_key: Optional[str] = None,
177        location_key: Optional[str] = None,
178        instance_keys: Optional[str] = None,
179        indexes: Union[Dict[str, str], List[str], None] = None,
180    ):
181        """
182        Parameters
183        ----------
184        connector: str
185            Keys for the pipe's source connector, e.g. `'sql:main'`.
186
187        metric: str
188            Label for the pipe's contents, e.g. `'weather'`.
189
190        location: str, default None
191            Label for the pipe's location. Defaults to `None`.
192
193        parameters: Optional[Dict[str, Any]], default None
194            Optionally set a pipe's parameters from the constructor,
195            e.g. columns and other attributes.
196            You can edit these parameters with `edit pipes`.
197
198        columns: Union[Dict[str, str], List[str], None], default None
199            Set the `columns` dictionary of `parameters`.
200            If `parameters` is also provided, this dictionary is added under the `'columns'` key.
201
202        indices: Optional[Dict[str, Union[str, List[str]]]], default None
203            Set the `indices` dictionary of `parameters`.
204            If `parameters` is also provided, this dictionary is added under the `'indices'` key.
205
206        tags: Optional[List[str]], default None
207            A list of strings to be added under the `'tags'` key of `parameters`.
208            You can select pipes with certain tags using `--tags`.
209
210        dtypes: Optional[Dict[str, str]], default None
211            Set the `dtypes` dictionary of `parameters`.
212            If `parameters` is also provided, this dictionary is added under the `'dtypes'` key.
213
214        mrsm_instance: Optional[Union[str, InstanceConnector]], default None
215            Connector for the Meerschaum instance where the pipe resides.
216            Defaults to the preconfigured default instance (`'sql:main'`).
217
218        instance: Optional[Union[str, InstanceConnector]], default None
219            Alias for `mrsm_instance`. If `mrsm_instance` is supplied, this value is ignored.
220
221        upsert: Optional[bool], default None
222            If `True`, set `upsert` to `True` in the parameters.
223
224        autoincrement: Optional[bool], default None
225            If `True`, set `autoincrement` in the parameters.
226
227        static: Optional[bool], default None
228            If `True`, set `static` in the parameters.
229
230        enforce: Optional[bool], default None
231            If `False`, skip data type enforcement.
232            Default behavior is `True`.
233
234        null_indices: Optional[bool], default None
235            Set to `False` if there will be no null values in the index columns.
236            Defaults to `True`.
237
238        temporary: bool, default False
239            If `True`, prevent instance tables (pipes, users, plugins) from being created.
240
241        cache: bool, default False
242            If `True`, cache fetched data into a local database file.
243            Defaults to `False`.
244        """
245        from meerschaum.utils.warnings import error, warn
246        if (not connector and not connector_keys) or (not metric and not metric_key):
247            error(
248                "Please provide strings for the connector and metric\n    "
249                + "(first two positional arguments)."
250            )
251
252        ### Fall back to legacy `location_key` just in case.
253        if not location:
254            location = location_key
255
256        if not connector:
257            connector = connector_keys
258
259        if not metric:
260            metric = metric_key
261
262        if location in ('[None]', 'None'):
263            location = None
264
265        from meerschaum.config.static import STATIC_CONFIG
266        negation_prefix = STATIC_CONFIG['system']['fetch_pipes_keys']['negation_prefix']
267        for k in (connector, metric, location, *(tags or [])):
268            if str(k).startswith(negation_prefix):
269                error(f"A pipe's keys and tags cannot start with the prefix '{negation_prefix}'.")
270
271        self.connector_keys = str(connector)
272        self.connector_key = self.connector_keys ### Alias
273        self.metric_key = metric
274        self.location_key = location
275        self.temporary = temporary
276
277        self._attributes = {
278            'connector_keys': self.connector_keys,
279            'metric_key': self.metric_key,
280            'location_key': self.location_key,
281            'parameters': {},
282        }
283
284        ### only set parameters if values are provided
285        if isinstance(parameters, dict):
286            self._attributes['parameters'] = parameters
287        else:
288            if parameters is not None:
289                warn(f"The provided parameters are of invalid type '{type(parameters)}'.")
290            self._attributes['parameters'] = {}
291
292        columns = columns or self._attributes.get('parameters', {}).get('columns', {})
293        if isinstance(columns, list):
294            columns = {str(col): str(col) for col in columns}
295        if isinstance(columns, dict):
296            self._attributes['parameters']['columns'] = columns
297        elif columns is not None:
298            warn(f"The provided columns are of invalid type '{type(columns)}'.")
299
300        indices = (
301            indices
302            or indexes
303            or self._attributes.get('parameters', {}).get('indices', None)
304            or self._attributes.get('parameters', {}).get('indexes', None)
305        )
306        if isinstance(indices, dict):
307            indices_key = (
308                'indexes'
309                if 'indexes' in self._attributes['parameters']
310                else 'indices'
311            )
312            self._attributes['parameters'][indices_key] = indices
313
314        if isinstance(tags, (list, tuple)):
315            self._attributes['parameters']['tags'] = tags
316        elif tags is not None:
317            warn(f"The provided tags are of invalid type '{type(tags)}'.")
318
319        if isinstance(target, str):
320            self._attributes['parameters']['target'] = target
321        elif target is not None:
322            warn(f"The provided target is of invalid type '{type(target)}'.")
323
324        if isinstance(dtypes, dict):
325            self._attributes['parameters']['dtypes'] = dtypes
326        elif dtypes is not None:
327            warn(f"The provided dtypes are of invalid type '{type(dtypes)}'.")
328
329        if isinstance(upsert, bool):
330            self._attributes['parameters']['upsert'] = upsert
331
332        if isinstance(autoincrement, bool):
333            self._attributes['parameters']['autoincrement'] = autoincrement
334
335        if isinstance(static, bool):
336            self._attributes['parameters']['static'] = static
337
338        if isinstance(enforce, bool):
339            self._attributes['parameters']['enforce'] = enforce
340
341        if isinstance(null_indices, bool):
342            self._attributes['parameters']['null_indices'] = null_indices
343
344        ### NOTE: The parameters dictionary is {} by default.
345        ###       A Pipe may be registered without parameters, then edited,
346        ###       or a Pipe may be registered with parameters set in-memory first.
347        _mrsm_instance = mrsm_instance if mrsm_instance is not None else (instance or instance_keys)
348        if _mrsm_instance is None:
349            _mrsm_instance = get_config('meerschaum', 'instance', patch=True)
350
351        if not isinstance(_mrsm_instance, str):
352            self._instance_connector = _mrsm_instance
353            self.instance_keys = str(_mrsm_instance)
354        else: ### NOTE: must be SQL or API Connector for this work
355            self.instance_keys = _mrsm_instance
356
357        self._cache = cache and get_config('system', 'experimental', 'cache')
358
359    @property
360    def meta(self):
361        """
362        Return the four keys needed to reconstruct this pipe.
363        """
364        return {
365            'connector_keys': self.connector_keys,
366            'metric_key': self.metric_key,
367            'location_key': self.location_key,
368            'instance_keys': self.instance_keys,
369        }
370
371    def keys(self) -> List[str]:
372        """
373        Return the ordered keys for this pipe.
374        """
375        return {
376            key: val
377            for key, val in self.meta.items()
378            if key != 'instance'
379        }
380
381    @property
382    def instance_connector(self) -> Union[InstanceConnector, None]:
383        """
384        The connector to where this pipe resides.
385        May either be of type `meerschaum.connectors.sql.SQLConnector` or
386        `meerschaum.connectors.api.APIConnector`.
387        """
388        if '_instance_connector' not in self.__dict__:
389            from meerschaum.connectors.parse import parse_instance_keys
390            conn = parse_instance_keys(self.instance_keys)
391            if conn:
392                self._instance_connector = conn
393            else:
394                return None
395        return self._instance_connector
396
397    @property
398    def connector(self) -> Union[meerschaum.connectors.Connector, None]:
399        """
400        The connector to the data source.
401        """
402        if '_connector' not in self.__dict__:
403            from meerschaum.connectors.parse import parse_instance_keys
404            import warnings
405            with warnings.catch_warnings():
406                warnings.simplefilter('ignore')
407                try:
408                    conn = parse_instance_keys(self.connector_keys)
409                except Exception:
410                    conn = None
411            if conn:
412                self._connector = conn
413            else:
414                return None
415        return self._connector
416
417    @property
418    def cache_connector(self) -> Union[meerschaum.connectors.sql.SQLConnector, None]:
419        """
420        If the pipe was created with `cache=True`, return the connector to the pipe's
421        SQLite database for caching.
422        """
423        if not self._cache:
424            return None
425
426        if '_cache_connector' not in self.__dict__:
427            from meerschaum.connectors import get_connector
428            from meerschaum.config._paths import DUCKDB_RESOURCES_PATH, SQLITE_RESOURCES_PATH
429            _resources_path = SQLITE_RESOURCES_PATH
430            self._cache_connector = get_connector(
431                'sql', '_cache_' + str(self),
432                flavor='sqlite',
433                database=str(_resources_path / ('_cache_' + str(self) + '.db')),
434            )
435
436        return self._cache_connector
437
438    @property
439    def cache_pipe(self) -> Union['meerschaum.Pipe', None]:
440        """
441        If the pipe was created with `cache=True`, return another `meerschaum.Pipe` used to
442        manage the local data.
443        """
444        if self.cache_connector is None:
445            return None
446        if '_cache_pipe' not in self.__dict__:
447            from meerschaum.config._patch import apply_patch_to_config
448            from meerschaum.utils.sql import sql_item_name
449            _parameters = copy.deepcopy(self.parameters)
450            _fetch_patch = {
451                'fetch': ({
452                    'definition': (
453                        "SELECT * FROM "
454                        + sql_item_name(
455                            str(self.target),
456                            self.instance_connector.flavor,
457                            self.instance_connector.get_pipe_schema(self),
458                        )
459                    ),
460                }) if self.instance_connector.type == 'sql' else ({
461                    'connector_keys': self.connector_keys,
462                    'metric_key': self.metric_key,
463                    'location_key': self.location_key,
464                })
465            }
466            _parameters = apply_patch_to_config(_parameters, _fetch_patch)
467            self._cache_pipe = Pipe(
468                self.instance_keys,
469                (self.connector_keys + '_' + self.metric_key + '_cache'),
470                self.location_key,
471                mrsm_instance = self.cache_connector,
472                parameters = _parameters,
473                cache = False,
474                temporary = True,
475            )
476
477        return self._cache_pipe
478
479    def __str__(self, ansi: bool=False):
480        return pipe_repr(self, ansi=ansi)
481
482    def __eq__(self, other):
483        try:
484            return (
485                isinstance(self, type(other))
486                and self.connector_keys == other.connector_keys
487                and self.metric_key == other.metric_key
488                and self.location_key == other.location_key
489                and self.instance_keys == other.instance_keys
490            )
491        except Exception:
492            return False
493
494    def __hash__(self):
495        ### Using an esoteric separator to avoid collisions.
496        sep = "[\"']"
497        return hash(
498            str(self.connector_keys) + sep
499            + str(self.metric_key) + sep
500            + str(self.location_key) + sep
501            + str(self.instance_keys) + sep
502        )
503
504    def __repr__(self, ansi: bool=True, **kw) -> str:
505        if not hasattr(sys, 'ps1'):
506            ansi = False
507
508        return pipe_repr(self, ansi=ansi, **kw)
509
510    def __pt_repr__(self):
511        from meerschaum.utils.packages import attempt_import
512        prompt_toolkit_formatted_text = attempt_import('prompt_toolkit.formatted_text', lazy=False)
513        return prompt_toolkit_formatted_text.ANSI(pipe_repr(self, ansi=True))
514
515    def __getstate__(self) -> Dict[str, Any]:
516        """
517        Define the state dictionary (pickling).
518        """
519        return {
520            'connector_keys': self.connector_keys,
521            'metric_key': self.metric_key,
522            'location_key': self.location_key,
523            'parameters': self.parameters,
524            'instance_keys': self.instance_keys,
525        }
526
527    def __setstate__(self, _state: Dict[str, Any]):
528        """
529        Read the state (unpickling).
530        """
531        self.__init__(**_state)
532
533    def __getitem__(self, key: str) -> Any:
534        """
535        Index the pipe's attributes.
536        If the `key` cannot be found`, return `None`.
537        """
538        if key in self.attributes:
539            return self.attributes.get(key, None)
540
541        aliases = {
542            'connector': 'connector_keys',
543            'connector_key': 'connector_keys',
544            'metric': 'metric_key',
545            'location': 'location_key',
546        }
547        aliased_key = aliases.get(key, None)
548        if aliased_key is not None:
549            return self.attributes.get(aliased_key, None)
550
551        property_aliases = {
552            'instance': 'instance_keys',
553            'instance_key': 'instance_keys',
554        }
555        aliased_key = property_aliases.get(key, None)
556        if aliased_key is not None:
557            key = aliased_key
558        return getattr(self, key, None)