meerschaum.connectors.api

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

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

188    @property
189    def session(self):
190        if self._session is None:
191            _ = attempt_import('certifi', lazy=False)
192            requests = attempt_import('requests', lazy=False)
193            if requests:
194                self._session = requests.Session()
195            if self._session is None:
196                error("Failed to import requests. Is requests installed?")
197        return self._session

199    @property
200    def token(self):
201        if self.login_scheme == 'api_key':
202            return self.api_key
203
204        expired = (
205            True if self._expires is None else (
206                (
207                    self._expires
208                    <
209                    datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(minutes=1)
210                )
211            )
212        )
213
214        if self._token is None or expired:
215            success, msg = self.login()
216            if not success and not self.__dict__.get('_emitted_warning'):
217                warn(msg, stack=False)
218                self._emitted_warning = True
219        return self._token

221    @property
222    def instance_keys(self) -> Union[str, None]:
223        """
224        Return the instance keys to be sent alongside pipe requests.
225        """
226        return self._instance_keys

228    @property
229    def login_scheme(self) -> str:
230        """
231        Return the login scheme to use based on the configured credentials.
232        """
233        if 'username' in self.__dict__:
234            return 'password'
235        if 'client_id' in self.__dict__:
236            return 'client_credentials'
237        elif 'api_key' in self.__dict__:
238            return 'api_key'
239
240        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) -> List[
117        Union[
118            Tuple[str, str, Union[str, None]],
119            Tuple[str, str, Union[str, None], List[str]],
120            Tuple[str, str, Union[str, None], Dict[str, Any]]
121        ]
122    ]:
123    """
124    Fetch registered Pipes' keys from the API.
125    
126    Parameters
127    ----------
128    connector_keys: Optional[List[str]], default None
129        The connector keys for the query.
130
131    metric_keys: Optional[List[str]], default None
132        The metric keys for the query.
133
134    location_keys: Optional[List[str]], default None
135        The location keys for the query.
136
137    tags: Optional[List[str]], default None
138        A list of tags for the query.
139
140    params: Optional[Dict[str, Any]], default None
141        A parameters dictionary for filtering against the `pipes` table
142        (e.g. `{'connector_keys': 'plugin:foo'}`).
143        Not recommeded to be used.
144
145    debug: bool, default False
146        Verbosity toggle.
147
148    Returns
149    -------
150    A list of tuples containing pipes' keys.
151    """
152    from meerschaum._internal.static import STATIC_CONFIG
153    if connector_keys is None:
154        connector_keys = []
155    if metric_keys is None:
156        metric_keys = []
157    if location_keys is None:
158        location_keys = []
159    if tags is None:
160        tags = []
161
162    r_url = STATIC_CONFIG['api']['endpoints']['pipes'] + '/keys'
163    try:
164        j = self.get(
165            r_url,
166            params={
167                'connector_keys': json.dumps(connector_keys),
168                'metric_keys': json.dumps(metric_keys),
169                'location_keys': json.dumps(location_keys),
170                'tags': json.dumps(tags),
171                'params': json.dumps(params),
172                'instance_keys': self.instance_keys,
173            },
174            debug=debug
175        ).json()
176    except Exception as e:
177        import traceback
178        traceback.print_exc()
179        error(str(e))
180
181    if 'detail' in j:
182        error(j['detail'], stack=False)
183    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

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