meerschaum.connectors.api

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

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

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

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

 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 register_pipe(
36    self,
37    pipe: mrsm.Pipe,
38    debug: bool = False
39) -> SuccessTuple:
40    """Submit a POST to the API to register a new Pipe object.
41    Returns a tuple of (success_bool, response_dict).
42    """
43    from meerschaum.utils.debug import dprint
44    ### NOTE: if `parameters` is supplied in the Pipe constructor,
45    ###       then `pipe.parameters` will exist and not be fetched from the database.
46    r_url = pipe_r_url(pipe)
47    response = self.post(
48        r_url + '/register',
49        json = pipe.parameters,
50        debug = debug,
51    )
52    if debug:
53        dprint(response.text)
54
55    if not response:
56        return False, response.text
57
58    response_data = response.json()
59    if isinstance(response_data, list):
60        response_tuple = response_data[0], response_data[1]
61    elif 'detail' in response.json():
62        response_tuple = response.__bool__(), response_data['detail']
63    else:
64        response_tuple = response.__bool__(), response.text
65    return response_tuple

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

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

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

316def delete_pipe(
317    self,
318    pipe: Optional[mrsm.Pipe] = None,
319    debug: bool = None,        
320) -> SuccessTuple:
321    """Delete a Pipe and drop its table."""
322    if pipe is None:
323        error("Pipe cannot be None.")
324    r_url = pipe_r_url(pipe)
325    response = self.delete(
326        r_url + '/delete',
327        debug = debug,
328    )
329    if debug:
330        dprint(response.text)
331
332    response_data = response.json()
333    if isinstance(response.json(), list):
334        response_tuple = response_data[0], response_data[1]
335    elif 'detail' in response.json():
336        response_tuple = response.__bool__(), response_data['detail']
337    else:
338        response_tuple = response.__bool__(), response.text
339    return response_tuple