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

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