meerschaum.connectors.api

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

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

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

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

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

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