meerschaum.connectors.api

126    def __init__(
127        self,
128        label: Optional[str] = None,
129        wait: bool = False,
130        debug: bool = False,
131        **kw
132    ):
133        if 'uri' in kw:
134            from_uri_params = self.from_uri(kw['uri'], as_dict=True)
135            label = label or from_uri_params.get('label', None)
136            _ = from_uri_params.pop('label', None)
137            kw.update(from_uri_params)
138
139        super().__init__('api', label=label, **kw)
140        if 'protocol' not in self.__dict__:
141            self.protocol = (
142                'https' if self.__dict__.get('uri', '').startswith('https')
143                else 'http'
144            )
145
146        if 'uri' not in self.__dict__:
147            self.verify_attributes(required_attributes)
148        else:
149            from meerschaum.connectors.sql import SQLConnector
150            conn_attrs = SQLConnector.parse_uri(self.__dict__['uri'])
151            if 'host' not in conn_attrs:
152                raise Exception(f"Invalid URI for '{self}'.")
153            self.__dict__.update(conn_attrs)
154
155        self.url = (
156            self.protocol + '://' +
157            self.host
158            + (
159                (':' + str(self.port))
160                if self.__dict__.get('port', None)
161                else ''
162            )
163        )
164        self._token = None
165        self._expires = None
166        self._session = None
167        self._instance_keys = self.__dict__.get('instance_keys', None)

170    @property
171    def URI(self) -> str:
172        """
173        Return the fully qualified URI.
174        """
175        import urllib.parse
176        username = self.__dict__.get('username', None)
177        password = self.__dict__.get('password', None)
178        client_id = self.__dict__.get('client_id', None)
179        client_secret = self.__dict__.get('client_secret', None)
180        api_key = self.__dict__.get('api_key', None)
181        creds = (username + ':' + password + '@') if username and password else ''
182        params = {}
183        params_str = ('?' + urllib.parse.urlencode(params)) if params else ''
184        return (
185            self.protocol
186            + '://'
187            + creds
188            + self.host
189            + (
190                (':' + str(self.port))
191                if self.__dict__.get('port', None)
192                else ''
193            )
194            + params_str
195        )

197    @property
198    def session(self):
199        if self._session is None:
200            _ = attempt_import('certifi', lazy=False)
201            requests = attempt_import('requests', lazy=False)
202            if requests:
203                self._session = requests.Session()
204            if self._session is None:
205                error("Failed to import requests. Is requests installed?")
206        return self._session

208    @property
209    def token(self):
210        if self.login_scheme == 'api_key':
211            return self.api_key
212
213        expired = (
214            True if self._expires is None else (
215                (
216                    self._expires
217                    <
218                    datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(minutes=1)
219                )
220            )
221        )
222
223        if self._token is None or expired:
224            success, msg = self.login()
225            if not success and not self.__dict__.get('_emitted_warning'):
226                warn(msg, stack=False)
227                self._emitted_warning = True
228        return self._token

230    @property
231    def instance_keys(self) -> Union[str, None]:
232        """
233        Return the instance keys to be sent alongside pipe requests.
234        """
235        return self._instance_keys

237    @property
238    def login_scheme(self) -> str:
239        """
240        Return the login scheme to use based on the configured credentials.
241        """
242        if 'username' in self.__dict__:
243            return 'password'
244        if 'client_id' in self.__dict__:
245            return 'client_credentials'
246        elif 'api_key' in self.__dict__:
247            return 'api_key'
248
249        return 'password'

 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(
50        job_name,
51        sysargs,
52        executor_keys=str(self),
53        ### Emit the entire output, not just the last `lines_to_show` (default 30) lines.
54        _properties={'logs': {'lines_to_show': 0}},
55    )
56
57    start_success, start_msg = job.start()
58    if not start_success:
59        return start_success, start_msg
60
61    await job.monitor_logs_async(
62        callback_function=callback_function,
63        stop_on_exit=True,
64        strip_timestamps=True,
65    )
66
67    success, msg = job.result
68    job.delete()
69    return success, msg

 72def do_action_legacy(
 73    self,
 74    action: Optional[List[str]] = None,
 75    sysargs: Optional[List[str]] = None,
 76    debug: bool = False,
 77    **kw
 78) -> SuccessTuple:
 79    """
 80    NOTE: This method is deprecated.
 81    Please use `do_action()` or `do_action_async()`.
 82
 83    Execute a Meerschaum action remotely.
 84
 85    If `sysargs` are provided, parse those instead.
 86    Otherwise infer everything from keyword arguments.
 87
 88    Examples
 89    --------
 90    >>> conn = mrsm.get_connector('api:main')
 91    >>> conn.do_action(['show', 'pipes'])
 92    (True, "Success")
 93    >>> conn.do_action(['show', 'arguments'], name='test')
 94    (True, "Success")
 95    """
 96    import sys, json
 97    from meerschaum.utils.debug import dprint
 98    from meerschaum._internal.static import STATIC_CONFIG
 99    from meerschaum.utils.misc import json_serialize_datetime
100    if action is None:
101        action = []
102
103    if sysargs is not None and action and action[0] == '':
104        from meerschaum._internal.arguments import parse_arguments
105        if debug:
106            dprint(f"Parsing sysargs:\n{sysargs}")
107        json_dict = parse_arguments(sysargs)
108    else:
109        json_dict = kw
110        json_dict['action'] = action
111        if 'noask' not in kw:
112            json_dict['noask'] = True
113        if 'yes' not in kw:
114            json_dict['yes'] = True
115        if debug:
116            json_dict['debug'] = debug
117
118    root_action = json_dict['action'][0]
119    del json_dict['action'][0]
120    r_url = f"{STATIC_CONFIG['api']['endpoints']['actions']}/{root_action}"
121    
122    if debug:
123        from meerschaum.utils.formatting import pprint
124        dprint(f"Sending data to '{self.url + r_url}':")
125        pprint(json_dict, stream=sys.stderr)
126
127    response = self.post(
128        r_url,
129        data = json.dumps(json_dict, default=json_serialize_datetime),
130        debug = debug,
131    )
132    try:
133        response_list = json.loads(response.text)
134        if isinstance(response_list, dict) and 'detail' in response_list:
135            return False, response_list['detail']
136    except Exception as e:
137        print(f"Invalid response: {response}")
138        print(e)
139        return False, response.text
140    if debug:
141        dprint(response)
142    try:
143        return response_list[0], response_list[1]
144    except Exception as e:
145        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._internal.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._internal.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[
117    Dict[Union[int, str], Tuple[str, str, Union[str, None], Dict[str, Any]]],
118    List[Tuple[str, str, Union[str, None]]],
119]:
120    """
121    Fetch registered Pipes' keys from the API.
122
123    Parameters
124    ----------
125    connector_keys: Optional[List[str]], default None
126        The connector keys for the query.
127
128    metric_keys: Optional[List[str]], default None
129        The metric keys for the query.
130
131    location_keys: Optional[List[str]], default None
132        The location keys for the query.
133
134    tags: Optional[List[str]], default None
135        A list of tags for the query.
136
137    params: Optional[Dict[str, Any]], default None
138        A parameters dictionary for filtering against the `pipes` table
139        (e.g. `{'connector_keys': 'plugin:foo'}`).
140        Not recommeded to be used.
141
142    debug: bool, default False
143        Verbosity toggle.
144
145    Returns
146    -------
147    A dictionary mapping pipe IDs to key tuples, or a list of key tuples for older servers.
148    """
149    from meerschaum._internal.static import STATIC_CONFIG
150    if connector_keys is None:
151        connector_keys = []
152    if metric_keys is None:
153        metric_keys = []
154    if location_keys is None:
155        location_keys = []
156    if tags is None:
157        tags = []
158
159    r_url = STATIC_CONFIG['api']['endpoints']['pipes'] + '/keys'
160    try:
161        j = self.get(
162            r_url,
163            params={
164                'connector_keys': json.dumps(connector_keys),
165                'metric_keys': json.dumps(metric_keys),
166                'location_keys': json.dumps(location_keys),
167                'tags': json.dumps(tags),
168                'params': json.dumps(params),
169                'instance_keys': self.instance_keys,
170                'as_dict': True,
171            },
172            debug=debug
173        ).json()
174    except Exception as e:
175        import traceback
176        traceback.print_exc()
177        error(str(e))
178
179    if 'detail' in j:
180        error(j['detail'], stack=False)
181
182    if isinstance(j, dict):
183        return {
184            (int(k) if str(k).isdigit() else k): tuple(v)
185            for k, v in j.items()
186        }
187    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.get_parameters(apply_symlinks=False),
 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

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