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    params: Optional[Dict[str, Any]] = None,
 35    mrsm_instance: Union[str, InstanceConnector, None] = None,
 36    instance: Union[str, InstanceConnector, None] = None,
 37    as_list: bool = False,
 38    as_tags_dict: bool = False,
 39    method: str = 'registered',
 40    workers: Optional[int] = None,
 41    debug: bool = False,
 42    _cache_parameters: bool = True,
 43    **kw: Any
 44) -> Union[PipesDict, List[mrsm.Pipe], Dict[str, mrsm.Pipe]]:
 45    """
 46    Return a dictionary or list of `meerschaum.Pipe` objects.
 47
 48    Parameters
 49    ----------
 50    connector_keys: Union[str, List[str], None], default None
 51        String or list of connector keys.
 52        If omitted or is `'*'`, fetch all possible keys.
 53        If a string begins with `'_'`, select keys that do NOT match the string.
 54
 55    metric_keys: Union[str, List[str], None], default None
 56        String or list of metric keys. See `connector_keys` for formatting.
 57
 58    location_keys: Union[str, List[str], None], default None
 59        String or list of location keys. See `connector_keys` for formatting.
 60
 61    tags: Optional[List[str]], default None
 62         If provided, only include pipes with these tags.
 63
 64    params: Optional[Dict[str, Any]], default None
 65        Dictionary of additional parameters to search by.
 66        Params are parsed into a SQL WHERE clause.
 67        E.g. `{'a': 1, 'b': 2}` equates to `'WHERE a = 1 AND b = 2'`
 68
 69    mrsm_instance: Union[str, InstanceConnector, None], default None
 70        Connector keys for the Meerschaum instance of the pipes.
 71        Must be a `meerschaum.connectors.sql.SQLConnector.SQLConnector` or
 72        `meerschaum.connectors.api.APIConnector.APIConnector`.
 73        
 74    as_list: bool, default False
 75        If `True`, return pipes in a list instead of a hierarchical dictionary.
 76        `False` : `{connector_keys: {metric_key: {location_key: Pipe}}}`
 77        `True`  : `[Pipe]`
 78
 79    as_tags_dict: bool, default False
 80        If `True`, return a dictionary mapping tags to pipes.
 81        Pipes with multiple tags will be repeated.
 82
 83    method: str, default 'registered'
 84        Available options: `['registered', 'explicit', 'all']`
 85        If `'registered'` (default), create pipes based on registered keys in the connector's pipes table
 86        (API or SQL connector, depends on mrsm_instance).
 87        If `'explicit'`, create pipes from provided connector_keys, metric_keys, and location_keys
 88        instead of consulting the pipes table. Useful for creating non-existent pipes.
 89        If `'all'`, create pipes from predefined metrics and locations. Required `connector_keys`.
 90        **NOTE:** Method `'all'` is not implemented!
 91
 92    workers: Optional[int], default None
 93        If provided (and `as_tags_dict` is `True`), set the number of workers for the pool
 94        to fetch tags.
 95        Only takes effect if the instance connector supports multi-threading
 96
 97    **kw: Any:
 98        Keyword arguments to pass to the `meerschaum.Pipe` constructor.
 99
100    Returns
101    -------
102    A dictionary of dictionaries and `meerschaum.Pipe` objects
103    in the connector, metric, location hierarchy.
104    If `as_list` is `True`, return a list of `meerschaum.Pipe` objects.
105    If `as_tags_dict` is `True`, return a dictionary mapping tags to pipes.
106
107    Examples
108    --------
109    ```
110    >>> ### Manual definition:
111    >>> pipes = {
112    ...     <connector_keys>: {
113    ...         <metric_key>: {
114    ...             <location_key>: Pipe(
115    ...                 <connector_keys>,
116    ...                 <metric_key>,
117    ...                 <location_key>,
118    ...             ),
119    ...         },
120    ...     },
121    ... },
122    >>> ### Accessing a single pipe:
123    >>> pipes['sql:main']['weather'][None]
124    >>> ### Return a list instead:
125    >>> get_pipes(as_list=True)
126    [Pipe('sql:main', 'weather')]
127    >>> get_pipes(as_tags_dict=True)
128    {'gvl': Pipe('sql:main', 'weather')}
129    ```
130    """
131
132    import json
133    from collections import defaultdict
134    from meerschaum.config import get_config
135    from meerschaum.utils.warnings import error
136    from meerschaum.utils.misc import filter_keywords
137    from meerschaum.utils.pool import get_pool
138
139    if connector_keys is None:
140        connector_keys = []
141    if metric_keys is None:
142        metric_keys = []
143    if location_keys is None:
144        location_keys = []
145    if params is None:
146        params = {}
147    if tags is None:
148        tags = []
149
150    if isinstance(connector_keys, str):
151        connector_keys = [connector_keys]
152    if isinstance(metric_keys, str):
153        metric_keys = [metric_keys]
154    if isinstance(location_keys, str):
155        location_keys = [location_keys]
156
157    ### Get SQL or API connector (keys come from `connector.fetch_pipes_keys()`).
158    if mrsm_instance is None:
159        mrsm_instance = instance
160    if mrsm_instance is None:
161        mrsm_instance = get_config('meerschaum', 'instance', patch=True)
162    if isinstance(mrsm_instance, str):
163        from meerschaum.connectors.parse import parse_instance_keys
164        connector = parse_instance_keys(keys=mrsm_instance, debug=debug)
165    else:
166        from meerschaum.connectors import instance_types
167        valid_connector = False
168        if hasattr(mrsm_instance, 'type'):
169            if mrsm_instance.type in instance_types:
170                valid_connector = True
171        if not valid_connector:
172            error(f"Invalid instance connector: {mrsm_instance}")
173        connector = mrsm_instance
174    if debug:
175        from meerschaum.utils.debug import dprint
176        dprint(f"Using instance connector: {connector}")
177    if not connector:
178        error(f"Could not create connector from keys: '{mrsm_instance}'")
179
180    ### Get a list of tuples for the keys needed to build pipes.
181    result = fetch_pipes_keys(
182        method,
183        connector,
184        connector_keys = connector_keys,
185        metric_keys = metric_keys,
186        location_keys = location_keys,
187        tags = tags,
188        params = params,
189        workers = workers,
190        debug = debug
191    )
192    if result is None:
193        error("Unable to build pipes!")
194
195    ### Populate the `pipes` dictionary with Pipes based on the keys
196    ### obtained from the chosen `method`.
197    from meerschaum import Pipe
198    pipes = {}
199    for keys_tuple in result:
200        ck, mk, lk = keys_tuple[0], keys_tuple[1], keys_tuple[2]
201        pipe_tags_or_parameters = keys_tuple[3] if len(keys_tuple) == 4 else None
202        pipe_parameters = (
203            pipe_tags_or_parameters
204            if isinstance(pipe_tags_or_parameters, (dict, str))
205            else None
206        )
207        if isinstance(pipe_parameters, str):
208            pipe_parameters = json.loads(pipe_parameters)
209        pipe_tags = (
210            pipe_tags_or_parameters
211            if isinstance(pipe_tags_or_parameters, list)
212            else (
213                pipe_tags_or_parameters.get('tags', [])
214                if isinstance(pipe_tags_or_parameters, dict)
215                else None
216            )
217        )
218
219        if ck not in pipes:
220            pipes[ck] = {}
221
222        if mk not in pipes[ck]:
223            pipes[ck][mk] = {}
224
225        pipe = Pipe(
226            ck, mk, lk,
227            mrsm_instance = connector,
228            parameters = pipe_parameters,
229            tags = pipe_tags,
230            debug = debug,
231            **filter_keywords(Pipe, **kw)
232        )
233        pipe.__dict__['_tags'] = pipe_tags
234        pipes[ck][mk][lk] = pipe
235
236    if not as_list and not as_tags_dict:
237        return pipes
238
239    from meerschaum.utils.misc import flatten_pipes_dict
240    pipes_list = flatten_pipes_dict(pipes)
241    if as_list:
242        return pipes_list
243
244    pool = get_pool(workers=(workers if connector.IS_THREAD_SAFE else 1))
245    def gather_pipe_tags(pipe: mrsm.Pipe) -> Tuple[mrsm.Pipe, List[str]]:
246        _tags = pipe.__dict__.get('_tags', None)
247        gathered_tags = _tags if _tags is not None else pipe.tags
248        return pipe, (gathered_tags or [])
249
250    tags_pipes = defaultdict(lambda: [])
251    pipes_tags = dict(pool.map(gather_pipe_tags, pipes_list))
252    for pipe, tags in pipes_tags.items():
253        for tag in (tags or []):
254            tags_pipes[tag].append(pipe)
255
256    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]

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

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