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)

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