meerschaum

 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)

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

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

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