29def get_pipes(
 30    connector_keys: Union[str, List[str], None] = None,
 31    metric_keys: Union[str, List[str], None] = None,
 32    location_keys: Union[str, List[str], None] = None,
 33    tags: Optional[List[str]] = None,
 34    datetime_dtypes: Optional[List[str]] = None,
 35    params: Optional[Dict[str, Any]] = None,
 36    mrsm_instance: Union[str, InstanceConnector, None] = None,
 37    instance: Union[str, InstanceConnector, None] = None,
 38    as_list: bool = False,
 39    as_tags_dict: bool = False,
 40    method: str = 'registered',
 41    workers: Optional[int] = None,
 42    debug: bool = False,
 43    _cache_parameters: bool = True,
 44    **kw: Any
 45) -> Union[PipesDict, List[mrsm.Pipe], Dict[str, mrsm.Pipe]]:
 46    """
 47    Return a dictionary or list of `meerschaum.Pipe` objects.
 48
 49    Parameters
 50    ----------
 51    connector_keys: Union[str, List[str], None], default None
 52        String or list of connector keys.
 53        If omitted or is `'*'`, fetch all possible keys.
 54        If a string begins with `'_'`, select keys that do NOT match the string.
 55
 56    metric_keys: Union[str, List[str], None], default None
 57        String or list of metric keys. See `connector_keys` for formatting.
 58
 59    location_keys: Union[str, List[str], None], default None
 60        String or list of location keys. See `connector_keys` for formatting.
 61
 62    tags: Optional[List[str]], default None
 63        If provided, only include pipes with these tags.
 64
 65    datetime_dtypes: Optional[List[str]], default None
 66        If provided, only include pipes with the corresponding `datetime` axis dtypes.
 67        Accepted values are `datetime`, `int`, `None` (or `null`, etc.).
 68        May be negated by `_`.
 69
 70    params: Optional[Dict[str, Any]], default None
 71        Dictionary of additional parameters to search by.
 72        Params are parsed into a SQL WHERE clause.
 73        E.g. `{'a': 1, 'b': 2}` equates to `'WHERE a = 1 AND b = 2'`
 74
 75    mrsm_instance: Union[str, InstanceConnector, None], default None
 76        Connector keys for the Meerschaum instance of the pipes.
 77        Must be a `meerschaum.connectors.sql.SQLConnector.SQLConnector` or
 78        `meerschaum.connectors.api.APIConnector.APIConnector`.
 79        
 80    as_list: bool, default False
 81        If `True`, return pipes in a list instead of a hierarchical dictionary.
 82        `False` : `{connector_keys: {metric_key: {location_key: Pipe}}}`
 83        `True`  : `[Pipe]`
 84
 85    as_tags_dict: bool, default False
 86        If `True`, return a dictionary mapping tags to pipes.
 87        Pipes with multiple tags will be repeated.
 88
 89    method: str, default 'registered'
 90        Available options: `['registered', 'explicit', 'all']`
 91        If `'registered'` (default), create pipes based on registered keys in the connector's pipes table
 92        (API or SQL connector, depends on mrsm_instance).
 93        If `'explicit'`, create pipes from provided connector_keys, metric_keys, and location_keys
 94        instead of consulting the pipes table. Useful for creating non-existent pipes.
 95        If `'all'`, create pipes from predefined metrics and locations. Required `connector_keys`.
 96        **NOTE:** Method `'all'` is not implemented!
 97
 98    workers: Optional[int], default None
 99        If provided (and `as_tags_dict` is `True`), set the number of workers for the pool
100        to fetch tags.
101        Only takes effect if the instance connector supports multi-threading
102
103    **kw: Any:
104        Keyword arguments to pass to the `meerschaum.Pipe` constructor.
105
106    Returns
107    -------
108    A dictionary of dictionaries and `meerschaum.Pipe` objects
109    in the connector, metric, location hierarchy.
110    If `as_list` is `True`, return a list of `meerschaum.Pipe` objects.
111    If `as_tags_dict` is `True`, return a dictionary mapping tags to pipes.
112
113    Examples
114    --------
115    ```
116    >>> ### Manual definition:
117    >>> pipes = {
118    ...     <connector_keys>: {
119    ...         <metric_key>: {
120    ...             <location_key>: Pipe(
121    ...                 <connector_keys>,
122    ...                 <metric_key>,
123    ...                 <location_key>,
124    ...             ),
125    ...         },
126    ...     },
127    ... },
128    >>> ### Accessing a single pipe:
129    >>> pipes['sql:main']['weather'][None]
130    >>> ### Return a list instead:
131    >>> get_pipes(as_list=True)
132    [Pipe('sql:main', 'weather')]
133    >>> get_pipes(as_tags_dict=True)
134    {'gvl': Pipe('sql:main', 'weather')}
135    ```
136    """
137    import json
138    from collections import defaultdict
139    from meerschaum.config import get_config
140    from meerschaum.config.static import STATIC_CONFIG
141    from meerschaum.utils.warnings import error
142    from meerschaum.utils.misc import filter_keywords, separate_negation_values
143    from meerschaum.utils.pool import get_pool
144    from meerschaum.utils.pipes import replace_pipes_syntax
145    from meerschaum.utils.debug import dprint
146    from meerschaum.utils.dtypes import value_is_null, get_current_timestamp
147    from meerschaum import Pipe
148
149    negation_prefix = STATIC_CONFIG['system']['fetch_pipes_keys']['negation_prefix']
150    if datetime_dtypes:
151        if isinstance(datetime_dtypes, str):
152            datetime_dtypes = [datetime_dtypes]
153        for _dt in datetime_dtypes:
154            _clean = str(_dt).lstrip(negation_prefix).lower()
155            if _clean not in ('datetime', 'int') and not value_is_null(_clean):
156                error(f"Invalid datetime dtype '{_dt}'.")
157
158    if connector_keys is None:
159        connector_keys = []
160    if metric_keys is None:
161        metric_keys = []
162    if location_keys is None:
163        location_keys = []
164    if params is None:
165        params = {}
166    if tags is None:
167        tags = []
168
169    if isinstance(connector_keys, str):
170        connector_keys = [connector_keys]
171    if isinstance(metric_keys, str):
172        metric_keys = [metric_keys]
173    if isinstance(location_keys, str):
174        location_keys = [location_keys]
175
176    ### Get SQL or API connector (keys come from `connector.fetch_pipes_keys()`).
177    if mrsm_instance is None:
178        mrsm_instance = instance
179    if mrsm_instance is None:
180        mrsm_instance = get_config('meerschaum', 'instance', patch=True)
181    if isinstance(mrsm_instance, str):
182        from meerschaum.connectors.parse import parse_instance_keys
183        connector = parse_instance_keys(keys=mrsm_instance, debug=debug)
184    else:
185        from meerschaum.connectors import instance_types
186        valid_connector = False
187        if hasattr(mrsm_instance, 'type'):
188            if mrsm_instance.type in instance_types:
189                valid_connector = True
190        if not valid_connector:
191            error(f"Invalid instance connector: {mrsm_instance}")
192        connector = mrsm_instance
193    if debug:
194        dprint(f"Using instance connector: {connector}")
195    if not connector:
196        error(f"Could not create connector from keys: '{mrsm_instance}'")
197
198    ### Get a list of tuples for the keys needed to build pipes.
199    result = fetch_pipes_keys(
200        method,
201        connector,
202        connector_keys = connector_keys,
203        metric_keys = metric_keys,
204        location_keys = location_keys,
205        tags = tags,
206        params = params,
207        workers = workers,
208        debug = debug
209    )
210    if result is None:
211        error("Unable to build pipes!")
212    result_items: List[Tuple] = (
213        list(result.items())
214        if isinstance(result, dict)
215        else [(None, keys_tuple) for keys_tuple in result]
216    )
217
218    ### Populate the `pipes` dictionary with Pipes based on the keys
219    ### obtained from the chosen `method`.
220    in_dtypes, ex_dtypes = separate_negation_values(datetime_dtypes or [])
221    pipes: PipesDict = {}
222    for pipe_id, keys_tuple in result_items:
223        ck, mk, lk = keys_tuple[0], keys_tuple[1], keys_tuple[2]
224        pipe_tags_or_parameters = keys_tuple[3] if len(keys_tuple) == 4 else None
225        pipe_parameters = (
226            pipe_tags_or_parameters
227            if isinstance(pipe_tags_or_parameters, (dict, str))
228            else None
229        )
230        if isinstance(pipe_parameters, str):
231            pipe_parameters = json.loads(pipe_parameters)
232        pipe_tags = (
233            pipe_tags_or_parameters
234            if isinstance(pipe_tags_or_parameters, list)
235            else (
236                pipe_tags_or_parameters.get('tags', [])
237                if isinstance(pipe_tags_or_parameters, dict)
238                else None
239            )
240        )
241
242        pipe = Pipe(
243            ck, mk, lk,
244            mrsm_instance = connector,
245            parameters = pipe_parameters,
246            tags = pipe_tags,
247            debug = debug,
248            **filter_keywords(Pipe, **kw)
249        )
250        pipe.__dict__['_tags'] = pipe_tags
251        if pipe_id is not None:
252            pipe._cache_value('_id', pipe_id, memory_only=True, debug=debug)
253        if pipe_parameters is not None:
254            now = get_current_timestamp('ms', as_int=True) / 1000
255            full_attributes = {
256                'connector_keys': ck,
257                'metric_key': mk,
258                'location_key': lk,
259                'parameters': pipe_parameters,
260            }
261            if pipe_id is not None:
262                full_attributes['pipe_id'] = pipe_id
263            pipe._cache_value('attributes', full_attributes, memory_only=True, debug=debug)
264            pipe._cache_value('_attributes_sync_time', now, memory_only=True, debug=debug)
265        if datetime_dtypes:
266            if pipe_parameters is None:
267                pipe_parameters = pipe.get_parameters(debug=debug)
268            columns_val = (pipe_parameters or {}).get('columns', {}) or {}
269            if isinstance(columns_val, str) and 'Pipe(' in columns_val:
270                columns_val = replace_pipes_syntax(columns_val)
271
272            dt_col = columns_val.get('datetime', None)
273            dt_typ = (
274                ((pipe_parameters or {}).get('dtypes', None) or {}).get(dt_col, None)
275                if dt_col
276                else None
277            )
278
279            def _dtype_matches(clean_d):
280                if not dt_col:
281                    return value_is_null(clean_d)
282                return (
283                    (clean_d == 'int' and 'int' in str(dt_typ).lower())
284                    or
285                    (clean_d == 'datetime' and 'int' not in str(dt_typ).lower())
286                )
287
288            in_match = not in_dtypes or any(_dtype_matches(d) for d in in_dtypes)
289            ex_match = bool(ex_dtypes and any(_dtype_matches(d) for d in ex_dtypes))
290            keep_pipe = in_match and not ex_match
291
292            if not keep_pipe:
293                continue
294
295        if ck not in pipes:
296            pipes[ck] = {}
297
298        if mk not in pipes[ck]:
299            pipes[ck][mk] = {}
300
301
302        pipes[ck][mk][lk] = pipe
303
304    if not as_list and not as_tags_dict:
305        return pipes
306
307    from meerschaum.utils.pipes import flatten_pipes_dict
308    pipes_list = flatten_pipes_dict(pipes)
309    if as_list:
310        return pipes_list
311
312    pool = get_pool(workers=(workers if connector.IS_THREAD_SAFE else 1))
313    def gather_pipe_tags(pipe: mrsm.Pipe) -> Tuple[mrsm.Pipe, List[str]]:
314        _tags = pipe.__dict__.get('_tags', None)
315        gathered_tags = _tags if _tags is not None else pipe.tags
316        return pipe, (gathered_tags or [])
317
318    tags_pipes = defaultdict(lambda: [])
319    pipes_tags = dict(pool.map(gather_pipe_tags, pipes_list))
320    for pipe, tags in pipes_tags.items():
321        for tag in (tags or []):
322            tags_pipes[tag].append(pipe)
323
324    return dict(tags_pipes)

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

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

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