meerschaum.connectors.api

109    def __init__(
110        self,
111        label: Optional[str] = None,
112        wait: bool = False,
113        debug: bool = False,
114        **kw
115    ):
116        if 'uri' in kw:
117            from_uri_params = self.from_uri(kw['uri'], as_dict=True)
118            label = label or from_uri_params.get('label', None)
119            _ = from_uri_params.pop('label', None)
120            kw.update(from_uri_params)
121
122        super().__init__('api', label=label, **kw)
123        if 'protocol' not in self.__dict__:
124            self.protocol = (
125                'https' if self.__dict__.get('uri', '').startswith('https')
126                else 'http'
127            )
128
129        if 'uri' not in self.__dict__:
130            self.verify_attributes(required_attributes)
131        else:
132            from meerschaum.connectors.sql import SQLConnector
133            conn_attrs = SQLConnector.parse_uri(self.__dict__['uri'])
134            if 'host' not in conn_attrs:
135                raise Exception(f"Invalid URI for '{self}'.")
136            self.__dict__.update(conn_attrs)
137
138        self.url = (
139            self.protocol + '://' +
140            self.host
141            + (
142                (':' + str(self.port))
143                if self.__dict__.get('port', None)
144                else ''
145            )
146        )
147        self._token = None
148        self._expires = None
149        self._session = None
150        self._instance_keys = self.__dict__.get('instance_keys', None)

153    @property
154    def URI(self) -> str:
155        """
156        Return the fully qualified URI.
157        """
158        username = self.__dict__.get('username', None)
159        password = self.__dict__.get('password', None)
160        creds = (username + ':' + password + '@') if username and password else ''
161        return (
162            self.protocol
163            + '://'
164            + creds
165            + self.host
166            + (
167                (':' + str(self.port))
168                if self.__dict__.get('port', None)
169                else ''
170            )
171        )

174    @property
175    def session(self):
176        if self._session is None:
177            _ = attempt_import('certifi', lazy=False)
178            requests = attempt_import('requests', lazy=False)
179            if requests:
180                self._session = requests.Session()
181            if self._session is None:
182                error("Failed to import requests. Is requests installed?")
183        return self._session

185    @property
186    def token(self):
187        expired = (
188            True if self._expires is None else (
189                (
190                    self._expires
191                    <
192                    datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(minutes=1)
193                )
194            )
195        )
196
197        if self._token is None or expired:
198            success, msg = self.login()
199            if not success and not self.__dict__.get('_emitted_warning'):
200                warn(msg, stack=False)
201                self._emitted_warning = True
202        return self._token

204    @property
205    def instance_keys(self) -> Union[str, None]:
206        """
207        Return the instance keys to be sent alongside pipe requests.
208        """
209        return self._instance_keys

 28def make_request(
 29    self,
 30    method: str,
 31    r_url: str,
 32    headers: Optional[Dict[str, Any]] = None,
 33    use_token: bool = True,
 34    debug: bool = False,
 35    **kwargs: Any
 36) -> 'requests.Response':
 37    """
 38    Make a request to this APIConnector's endpoint using the in-memory session.
 39
 40    Parameters
 41    ----------
 42    method: str
 43        The kind of request to make.
 44        Accepted values:
 45        - `'GET'`
 46        - `'OPTIONS'`
 47        - `'HEAD'`
 48        - `'POST'`
 49        - `'PUT'`
 50        - `'PATCH'`
 51        - `'DELETE'`
 52
 53    r_url: str
 54        The relative URL for the endpoint (e.g. `'/pipes'`).
 55
 56    headers: Optional[Dict[str, Any]], default None
 57        The headers to use for the request.
 58        If `use_token` is `True`, the authorization token will be added to a copy of these headers.
 59
 60    use_token: bool, default True
 61        If `True`, add the authorization token to the headers.
 62
 63    debug: bool, default False
 64        Verbosity toggle.
 65
 66    kwargs: Any
 67        All other keyword arguments are passed to `requests.request`.
 68
 69    Returns
 70    -------
 71    A `requests.Reponse` object.
 72    """
 73    if method.upper() not in METHODS:
 74        raise ValueError(f"Method '{method}' is not supported.")
 75
 76    verify = self.__dict__.get('verify', None)
 77    if 'verify' not in kwargs and isinstance(verify, bool):
 78        kwargs['verify'] = verify
 79
 80    headers = (
 81        copy.deepcopy(headers)
 82        if isinstance(headers, dict)
 83        else {}
 84    )
 85
 86    if use_token:
 87        headers.update({'Authorization': f'Bearer {self.token}'})
 88
 89    if 'timeout' not in kwargs:
 90        kwargs['timeout'] = STATIC_CONFIG['api']['default_timeout']
 91
 92    request_url = urllib.parse.urljoin(self.url, r_url)
 93    if debug:
 94        dprint(f"[{self}] Sending a '{method.upper()}' request to {request_url}")
 95
 96    return self.session.request(
 97        method.upper(),
 98        request_url,
 99        headers=headers,
100        **kwargs
101    )

104def get(self, r_url: str, **kwargs: Any) -> 'requests.Response':
105    """
106    Wrapper for `requests.get`.
107
108    Parameters
109    ----------
110    r_url: str
111        The relative URL for the endpoint (e.g. `'/pipes'`).
112
113    headers: Optional[Dict[str, Any]], default None
114        The headers to use for the request.
115        If `use_token` is `True`, the authorization token will be added to a copy of these headers.
116
117    use_token: bool, default True
118        If `True`, add the authorization token to the headers.
119
120    debug: bool, default False
121        Verbosity toggle.
122
123    kwargs: Any
124        All other keyword arguments are passed to `requests.request`.
125
126    Returns
127    -------
128    A `requests.Reponse` object.
129
130    """
131    return self.make_request('GET', r_url, **kwargs)

134def post(self, r_url: str, **kwargs: Any) -> 'requests.Response':
135    """
136    Wrapper for `requests.post`.
137
138    Parameters
139    ----------
140    r_url: str
141        The relative URL for the endpoint (e.g. `'/pipes'`).
142
143    headers: Optional[Dict[str, Any]], default None
144        The headers to use for the request.
145        If `use_token` is `True`, the authorization token will be added to a copy of these headers.
146
147    use_token: bool, default True
148        If `True`, add the authorization token to the headers.
149
150    debug: bool, default False
151        Verbosity toggle.
152
153    kwargs: Any
154        All other keyword arguments are passed to `requests.request`.
155
156    Returns
157    -------
158    A `requests.Reponse` object.
159
160    """
161    return self.make_request('POST', r_url, **kwargs)

193def put(self, r_url: str, **kwargs: Any) -> 'requests.Response':
194    """
195    Wrapper for `requests.put`.
196
197    Parameters
198    ----------
199    r_url: str
200        The relative URL for the endpoint (e.g. `'/pipes'`).
201
202    headers: Optional[Dict[str, Any]], default None
203        The headers to use for the request.
204        If `use_token` is `True`, the authorization token will be added to a copy of these headers.
205
206    use_token: bool, default True
207        If `True`, add the authorization token to the headers.
208
209    debug: bool, default False
210        Verbosity toggle.
211
212    kwargs: Any
213        All other keyword arguments are passed to `requests.request`.
214
215    Returns
216    -------
217    A `requests.Reponse` object.
218    """
219    return self.make_request('PUT', r_url, **kwargs)

164def patch(self, r_url: str, **kwargs: Any) -> 'requests.Response':
165    """
166    Wrapper for `requests.patch`.
167
168    Parameters
169    ----------
170    r_url: str
171        The relative URL for the endpoint (e.g. `'/pipes'`).
172
173    headers: Optional[Dict[str, Any]], default None
174        The headers to use for the request.
175        If `use_token` is `True`, the authorization token will be added to a copy of these headers.
176
177    use_token: bool, default True
178        If `True`, add the authorization token to the headers.
179
180    debug: bool, default False
181        Verbosity toggle.
182
183    kwargs: Any
184        All other keyword arguments are passed to `requests.request`.
185
186    Returns
187    -------
188    A `requests.Reponse` object.
189    """
190    return self.make_request('PATCH', r_url, **kwargs)

222def delete(self, r_url: str, **kwargs: Any) -> 'requests.Response':
223    """
224    Wrapper for `requests.delete`.
225
226    Parameters
227    ----------
228    r_url: str
229        The relative URL for the endpoint (e.g. `'/pipes'`).
230
231    headers: Optional[Dict[str, Any]], default None
232        The headers to use for the request.
233        If `use_token` is `True`, the authorization token will be added to a copy of these headers.
234
235    use_token: bool, default True
236        If `True`, add the authorization token to the headers.
237
238    debug: bool, default False
239        Verbosity toggle.
240
241    kwargs: Any
242        All other keyword arguments are passed to `requests.request`.
243
244    Returns
245    -------
246    A `requests.Reponse` object.
247    """
248    return self.make_request('DELETE', r_url, **kwargs)

251def wget(
252    self,
253    r_url: str,
254    dest: Optional[Union[str, pathlib.Path]] = None,
255    headers: Optional[Dict[str, Any]] = None,
256    use_token: bool = True,
257    debug: bool = False,
258    **kw: Any
259) -> pathlib.Path:
260    """Mimic wget with requests."""
261    from meerschaum.utils.misc import wget
262    if headers is None:
263        headers = {}
264    if use_token:
265        headers.update({'Authorization': f'Bearer {self.token}'})
266    request_url = urllib.parse.urljoin(self.url, r_url)
267    if debug:
268        dprint(
269            f"[{self}] Downloading {request_url}"
270            + (f' to {dest}' if dest is not None else '')
271            + "..."
272        )
273    return wget(request_url, dest=dest, headers=headers, **kw)

24def get_actions(self):
25    """Get available actions from the API instance."""
26    return self.get(ACTIONS_ENDPOINT)

29def do_action(self, sysargs: List[str]) -> SuccessTuple:
30    """
31    Execute a Meerschaum action remotely.
32    """
33    return asyncio.run(self.do_action_async(sysargs))

36async def do_action_async(
37    self,
38    sysargs: List[str],
39    callback_function: Callable[[str], None] = partial(print, end=''),
40) -> SuccessTuple:
41    """
42    Execute an action as a temporary remote job.
43    """
44    from meerschaum._internal.arguments import remove_api_executor_keys
45    from meerschaum.utils.misc import generate_password
46    sysargs = remove_api_executor_keys(sysargs)
47
48    job_name = TEMP_PREFIX + generate_password(12)
49    job = mrsm.Job(job_name, sysargs, executor_keys=str(self))
50
51    start_success, start_msg = job.start()
52    if not start_success:
53        return start_success, start_msg
54
55    await job.monitor_logs_async(
56        callback_function=callback_function,
57        stop_on_exit=True,
58        strip_timestamps=True,
59    )
60
61    success, msg = job.result
62    job.delete()
63    return success, msg

 66def do_action_legacy(
 67    self,
 68    action: Optional[List[str]] = None,
 69    sysargs: Optional[List[str]] = None,
 70    debug: bool = False,
 71    **kw
 72) -> SuccessTuple:
 73    """
 74    NOTE: This method is deprecated.
 75    Please use `do_action()` or `do_action_async()`.
 76
 77    Execute a Meerschaum action remotely.
 78
 79    If `sysargs` are provided, parse those instead.
 80    Otherwise infer everything from keyword arguments.
 81
 82    Examples
 83    --------
 84    >>> conn = mrsm.get_connector('api:main')
 85    >>> conn.do_action(['show', 'pipes'])
 86    (True, "Success")
 87    >>> conn.do_action(['show', 'arguments'], name='test')
 88    (True, "Success")
 89    """
 90    import sys, json
 91    from meerschaum.utils.debug import dprint
 92    from meerschaum.config.static import STATIC_CONFIG
 93    from meerschaum.utils.misc import json_serialize_datetime
 94    if action is None:
 95        action = []
 96
 97    if sysargs is not None and action and action[0] == '':
 98        from meerschaum._internal.arguments import parse_arguments
 99        if debug:
100            dprint(f"Parsing sysargs:\n{sysargs}")
101        json_dict = parse_arguments(sysargs)
102    else:
103        json_dict = kw
104        json_dict['action'] = action
105        if 'noask' not in kw:
106            json_dict['noask'] = True
107        if 'yes' not in kw:
108            json_dict['yes'] = True
109        if debug:
110            json_dict['debug'] = debug
111
112    root_action = json_dict['action'][0]
113    del json_dict['action'][0]
114    r_url = f"{STATIC_CONFIG['api']['endpoints']['actions']}/{root_action}"
115    
116    if debug:
117        from meerschaum.utils.formatting import pprint
118        dprint(f"Sending data to '{self.url + r_url}':")
119        pprint(json_dict, stream=sys.stderr)
120
121    response = self.post(
122        r_url,
123        data = json.dumps(json_dict, default=json_serialize_datetime),
124        debug = debug,
125    )
126    try:
127        response_list = json.loads(response.text)
128        if isinstance(response_list, dict) and 'detail' in response_list:
129            return False, response_list['detail']
130    except Exception as e:
131        print(f"Invalid response: {response}")
132        print(e)
133        return False, response.text
134    if debug:
135        dprint(response)
136    try:
137        return response_list[0], response_list[1]
138    except Exception as e:
139        return False, f"Failed to parse result from action '{root_action}'"

13def get_mrsm_version(self, **kw) -> Optional[str]:
14    """
15    Return the Meerschaum version of the API instance.
16    """
17    from meerschaum.config.static import STATIC_CONFIG
18    try:
19        j = self.get(
20            STATIC_CONFIG['api']['endpoints']['version'] + '/mrsm',
21            use_token=False,
22            **kw
23        ).json()
24    except Exception:
25        return None
26    if isinstance(j, dict) and 'detail' in j:
27        return None
28    return j

31def get_chaining_status(self, **kw) -> Optional[bool]:
32    """
33    Fetch the chaining status of the API instance.
34    """
35    from meerschaum.config.static import STATIC_CONFIG
36    try:
37        response = self.get(
38            STATIC_CONFIG['api']['endpoints']['chaining'],
39            use_token = True,
40            **kw
41        )
42        if not response:
43            return None
44    except Exception:
45        return None
46
47    return response.json()

35def get_pipe_instance_keys(self, pipe: mrsm.Pipe) -> Union[str, None]:
36    """
37    Return the configured instance keys for a pipe if set,
38    else fall back to the default `instance_keys` for this `APIConnector`.
39    """
40    return pipe.parameters.get('instance_keys', self.instance_keys)

43def register_pipe(
44    self,
45    pipe: mrsm.Pipe,
46    debug: bool = False
47) -> SuccessTuple:
48    """Submit a POST to the API to register a new Pipe object.
49    Returns a tuple of (success_bool, response_dict).
50    """
51    from meerschaum.utils.debug import dprint
52    r_url = pipe_r_url(pipe)
53    response = self.post(
54        r_url + '/register',
55        json=pipe._attributes.get('parameters', {}),
56        params={'instance_keys': self.get_pipe_instance_keys(pipe)},
57        debug=debug,
58    )
59    if debug:
60        dprint(response.text)
61
62    if not response:
63        return False, response.text
64
65    response_data = response.json()
66    if isinstance(response_data, list):
67        response_tuple = response_data[0], response_data[1]
68    elif 'detail' in response.json():
69        response_tuple = response.__bool__(), response_data['detail']
70    else:
71        response_tuple = response.__bool__(), response.text
72    return response_tuple

108def fetch_pipes_keys(
109    self,
110    connector_keys: Optional[List[str]] = None,
111    metric_keys: Optional[List[str]] = None,
112    location_keys: Optional[List[str]] = None,
113    tags: Optional[List[str]] = None,
114    params: Optional[Dict[str, Any]] = None,
115    debug: bool = False
116) -> Union[List[Tuple[str, str, Union[str, None]]]]:
117    """
118    Fetch registered Pipes' keys from the API.
119    
120    Parameters
121    ----------
122    connector_keys: Optional[List[str]], default None
123        The connector keys for the query.
124
125    metric_keys: Optional[List[str]], default None
126        The metric keys for the query.
127
128    location_keys: Optional[List[str]], default None
129        The location keys for the query.
130
131    tags: Optional[List[str]], default None
132        A list of tags for the query.
133
134    params: Optional[Dict[str, Any]], default None
135        A parameters dictionary for filtering against the `pipes` table
136        (e.g. `{'connector_keys': 'plugin:foo'}`).
137        Not recommeded to be used.
138
139    debug: bool, default False
140        Verbosity toggle.
141
142    Returns
143    -------
144    A list of tuples containing pipes' keys.
145    """
146    from meerschaum.config.static import STATIC_CONFIG
147    if connector_keys is None:
148        connector_keys = []
149    if metric_keys is None:
150        metric_keys = []
151    if location_keys is None:
152        location_keys = []
153    if tags is None:
154        tags = []
155
156    r_url = STATIC_CONFIG['api']['endpoints']['pipes'] + '/keys'
157    try:
158        j = self.get(
159            r_url,
160            params={
161                'connector_keys': json.dumps(connector_keys),
162                'metric_keys': json.dumps(metric_keys),
163                'location_keys': json.dumps(location_keys),
164                'tags': json.dumps(tags),
165                'params': json.dumps(params),
166                'instance_keys': self.instance_keys,
167            },
168            debug=debug
169        ).json()
170    except Exception as e:
171        error(str(e))
172
173    if 'detail' in j:
174        error(j['detail'], stack=False)
175    return [tuple(r) for r in j]

 75def edit_pipe(
 76    self,
 77    pipe: mrsm.Pipe,
 78    patch: bool = False,
 79    debug: bool = False,
 80) -> SuccessTuple:
 81    """Submit a PATCH to the API to edit an existing Pipe object.
 82    Returns a tuple of (success_bool, response_dict).
 83    """
 84    from meerschaum.utils.debug import dprint
 85    ### NOTE: if `parameters` is supplied in the Pipe constructor,
 86    ###       then `pipe.parameters` will exist and not be fetched from the database.
 87    r_url = pipe_r_url(pipe)
 88    response = self.patch(
 89        r_url + '/edit',
 90        params={'patch': patch, 'instance_keys': self.get_pipe_instance_keys(pipe)},
 91        json=pipe.parameters,
 92        debug=debug,
 93    )
 94    if debug:
 95        dprint(response.text)
 96
 97    response_data = response.json()
 98
 99    if isinstance(response.json(), list):
100        response_tuple = response_data[0], response_data[1]
101    elif 'detail' in response.json():
102        response_tuple = response.__bool__(), response_data['detail']
103    else:
104        response_tuple = response.__bool__(), response.text
105    return response_tuple

178def sync_pipe(
179    self,
180    pipe: mrsm.Pipe,
181    df: Optional[Union['pd.DataFrame', Dict[Any, Any], str]] = None,
182    chunksize: Optional[int] = -1,
183    debug: bool = False,
184    **kw: Any
185) -> SuccessTuple:
186    """Sync a DataFrame into a Pipe."""
187    from decimal import Decimal
188    from meerschaum.utils.debug import dprint
189    from meerschaum.utils.misc import json_serialize_datetime, items_str, interval_str
190    from meerschaum.config import get_config
191    from meerschaum.utils.packages import attempt_import
192    from meerschaum.utils.dataframe import get_numeric_cols, to_json
193    begin = time.perf_counter()
194    more_itertools = attempt_import('more_itertools')
195    if df is None:
196        msg = f"DataFrame is `None`. Cannot sync {pipe}."
197        return False, msg
198
199    def get_json_str(c):
200        ### allow syncing dict or JSON without needing to import pandas (for IOT devices)
201        if isinstance(c, (dict, list)):
202            return json.dumps(c, default=json_serialize_datetime)
203        return to_json(c, orient='columns')
204
205    df = json.loads(df) if isinstance(df, str) else df
206
207    _chunksize: Optional[int] = (1 if chunksize is None else (
208        get_config('system', 'connectors', 'sql', 'chunksize') if chunksize == -1
209        else chunksize
210    ))
211    keys: List[str] = list(df.columns)
212    chunks = []
213    if hasattr(df, 'index'):
214        df = df.reset_index(drop=True)
215        is_dask = 'dask' in df.__module__
216        chunks = (
217            (df.iloc[i] for i in more_itertools.chunked(df.index, _chunksize))
218            if not is_dask
219            else [partition.compute() for partition in df.partitions]
220        )
221
222        numeric_cols = get_numeric_cols(df)
223        if numeric_cols:
224            for col in numeric_cols:
225                df[col] = df[col].apply(lambda x: f'{x:f}' if isinstance(x, Decimal) else x)
226            pipe_dtypes = pipe.dtypes
227            new_numeric_cols = [
228                col
229                for col in numeric_cols
230                if pipe_dtypes.get(col, None) != 'numeric'
231            ]
232            pipe.dtypes.update({
233                col: 'numeric'
234                for col in new_numeric_cols
235            })
236            edit_success, edit_msg = pipe.edit(debug=debug)
237            if not edit_success:
238                warn(
239                    "Failed to update new numeric columns "
240                    + f"{items_str(new_numeric_cols)}:\n{edit_msg}"
241                )
242    elif isinstance(df, dict):
243        ### `_chunks` is a dict of lists of dicts.
244        ### e.g. {'a' : [ {'a':[1, 2]}, {'a':[3, 4]} ] }
245        _chunks = {k: [] for k in keys}
246        for k in keys:
247            chunk_iter = more_itertools.chunked(df[k], _chunksize)
248            for l in chunk_iter:
249                _chunks[k].append({k: l})
250
251        ### `chunks` is a list of dicts (e.g. orient by rows in pandas JSON).
252        for k, l in _chunks.items():
253            for i, c in enumerate(l):
254                try:
255                    chunks[i].update(c)
256                except IndexError:
257                    chunks.append(c)
258    elif isinstance(df, list):
259        chunks = (df[i] for i in more_itertools.chunked(df, _chunksize))
260
261    ### Send columns in case the user has defined them locally.
262    request_params = kw.copy()
263    if pipe.columns:
264        request_params['columns'] = json.dumps(pipe.columns)
265    request_params['instance_keys'] = self.get_pipe_instance_keys(pipe)
266    r_url = pipe_r_url(pipe) + '/data'
267
268    rowcount = 0
269    num_success_chunks = 0
270    for i, c in enumerate(chunks):
271        if debug:
272            dprint(f"[{self}] Posting chunk {i} to {r_url}...")
273        if len(c) == 0:
274            if debug:
275                dprint(f"[{self}] Skipping empty chunk...")
276            continue
277        json_str = get_json_str(c)
278
279        try:
280            response = self.post(
281                r_url,
282                params=request_params,
283                data=json_str,
284                debug=debug,
285            )
286        except Exception as e:
287            msg = f"Failed to post a chunk to {pipe}:\n{e}"
288            warn(msg)
289            return False, msg
290            
291        if not response:
292            return False, f"Failed to sync a chunk:\n{response.text}"
293
294        try:
295            j = json.loads(response.text)
296        except Exception as e:
297            return False, f"Failed to parse response from syncing {pipe}:\n{e}"
298
299        if isinstance(j, dict) and 'detail' in j:
300            return False, j['detail']
301
302        try:
303            j = tuple(j)
304        except Exception:
305            return False, response.text
306
307        if debug:
308            dprint("Received response: " + str(j))
309        if not j[0]:
310            return j
311
312        rowcount += len(c)
313        num_success_chunks += 1
314
315    success_tuple = True, (
316        f"It took {interval_str(timedelta(seconds=(time.perf_counter() - begin)))} "
317        + "to sync {rowcount:,} row"
318        + ('s' if rowcount != 1 else '')
319        + f" across {num_success_chunks:,} chunk" + ('s' if num_success_chunks != 1 else '') +
320        f" to {pipe}."
321    )
322    return success_tuple

325def delete_pipe(
326    self,
327    pipe: Optional[mrsm.Pipe] = None,
328    debug: bool = None,        
329) -> SuccessTuple:
330    """Delete a Pipe and drop its table."""
331    if pipe is None:
332        error("Pipe cannot be None.")
333    r_url = pipe_r_url(pipe)
334    response = self.delete(
335        r_url + '/delete',
336        params={'instance_keys': self.get_pipe_instance_keys(pipe)},
337        debug=debug,
338    )
339    if debug:
340        dprint(response.text)
341
342    response_data = response.json()
343    if isinstance(response.json(), list):
344        response_tuple = response_data[0], response_data[1]
345    elif 'detail' in response.json():
346        response_tuple = response.__bool__(), response_data['detail']
347    else:
348        response_tuple = response.__bool__(), response.text
349    return response_tuple