meerschaum.connectors.api

144    def __init__(
145        self,
146        label: Optional[str] = None,
147        wait: bool = False,
148        debug: bool = False,
149        **kw
150    ):
151        if 'uri' in kw:
152            from_uri_params = self.from_uri(kw['uri'], as_dict=True)
153            label = label or from_uri_params.get('label', None)
154            _ = from_uri_params.pop('label', None)
155            kw.update(from_uri_params)
156
157        super().__init__('api', label=label, **kw)
158        if 'protocol' not in self.__dict__:
159            self.protocol = (
160                'https' if self.__dict__.get('uri', '').startswith('https')
161                else 'http'
162            )
163
164        if 'uri' not in self.__dict__:
165            self.verify_attributes(required_attributes)
166        else:
167            conn_attrs = self.from_uri(self.__dict__['uri'], as_dict=True)
168            self.__dict__.update({
169                key: val
170                for key, val in conn_attrs.items()
171                if key not in ('label', 'protocol')
172            })
173
174        self.url = (
175            self.protocol + '://' +
176            self.host
177            + (
178                (':' + str(self.port))
179                if self.__dict__.get('port', None)
180                else ''
181            )
182        )
183        self._rust_client_obj = None
184        self._instance_keys = self.__dict__.get('instance_keys', None)

225    @property
226    def URI(self) -> str:
227        """
228        Return the fully qualified URI.
229        """
230        return self._rust_client().uri

232    @property
233    def session(self):
234        """
235        A `requests.Session` (not used by the connector, which makes its requests in Rust).
236        It exists for plugins which still reach for it and needs `requests` installed.
237        """
238        if self.__dict__.get('_session', None) is None:
239            from meerschaum.utils.packages import attempt_import
240            requests = attempt_import('requests', lazy=False)
241            if requests is None:
242                raise ImportError("Failed to import requests. Is requests installed?")
243            self._session = requests.Session()
244        return self._session

246    @property
247    def token(self):
248        """
249        Return the API key, or the session token (logging in when it is missing or expiring).
250        """
251        return self._rust_client().token

290    @property
291    def instance_keys(self) -> Union[str, None]:
292        """
293        Return the instance keys to be sent alongside pipe requests.
294        """
295        return self._instance_keys

297    @property
298    def login_scheme(self) -> str:
299        """
300        Return the login scheme to use based on the configured credentials.
301        """
302        return self._rust_client().login_scheme

21def make_request(
22    self,
23    method: str,
24    r_url: str,
25    headers: Optional[Dict[str, Any]] = None,
26    use_token: bool = True,
27    debug: bool = False,
28    **kwargs: Any
29) -> 'meerschaum._core.ApiResponse':
30    """
31    Make a request to the API and return a response object
32    (`status_code`, `ok`, `text`, `content`, `headers`, `json()`, `raise_for_status()`).
33
34    Parameters
35    ----------
36    method: str
37        The name of the request method (e.g. `'GET'`, `'POST'`).
38
39    r_url: str
40        The relative URL for the request (e.g. `'/pipes'`).
41
42    headers: Optional[Dict[str, Any]], default None
43        Headers to override the default headers.
44
45    use_token: bool, default True
46        If `True`, add the authorization token to the headers.
47
48    debug: bool, default False
49        Verbosity toggle.
50
51    kwargs: Any
52        `params` (the query), `json`, `data`, `files` and `timeout` are sent as `requests` would.
53    """
54    if method.upper() not in METHODS:
55        raise ValueError(f"Method '{method}' is not supported.")
56
57    if debug:
58        dprint(f"[{self}] Sending a '{method.upper()}' request to {self.url}{r_url}")
59
60    return self._rust_client().make_request(
61        method,
62        r_url,
63        headers=headers,
64        use_token=use_token,
65        **kwargs
66    )

69def get(self, r_url: str, **kwargs: Any) -> 'meerschaum._core.ApiResponse':
70    """
71    Send a `GET` request (see `make_request()`).
72    """
73    return self.make_request('GET', r_url, **kwargs)

76def post(self, r_url: str, **kwargs: Any) -> 'meerschaum._core.ApiResponse':
77    """
78    Send a `POST` request (see `make_request()`).
79    """
80    return self.make_request('POST', r_url, **kwargs)

90def put(self, r_url: str, **kwargs: Any) -> 'meerschaum._core.ApiResponse':
91    """
92    Send a `PUT` request (see `make_request()`).
93    """
94    return self.make_request('PUT', r_url, **kwargs)

83def patch(self, r_url: str, **kwargs: Any) -> 'meerschaum._core.ApiResponse':
84    """
85    Send a `PATCH` request (see `make_request()`).
86    """
87    return self.make_request('PATCH', r_url, **kwargs)

 97def delete(self, r_url: str, **kwargs: Any) -> 'meerschaum._core.ApiResponse':
 98    """
 99    Send a `DELETE` request (see `make_request()`).
100    """
101    return self.make_request('DELETE', r_url, **kwargs)

104def wget(
105    self,
106    r_url: str,
107    dest: Optional[Union[str, pathlib.Path]] = None,
108    headers: Optional[Dict[str, Any]] = None,
109    use_token: bool = True,
110    debug: bool = False,
111    **kw: Any
112) -> pathlib.Path:
113    """Mimic wget: download a resource of the API (with the token) to `dest`."""
114    from meerschaum.utils.misc import wget
115    if headers is None:
116        headers = {}
117    if use_token:
118        headers.update({'Authorization': f'Bearer {self.token}'})
119    request_url = urllib.parse.urljoin(self.url, r_url)
120    if debug:
121        dprint(
122            f"[{self}] Downloading {request_url}"
123            + (f' to {dest}' if dest is not None else '')
124            + "..."
125        )
126    return wget(request_url, dest=dest, headers=headers, **kw)

18def get_actions(self):
19    """
20    Get available actions from the API instance.
21    Returns the response (v4 does: `.json()` is the list of action names).
22    """
23    from meerschaum._internal.static import STATIC_CONFIG
24    return self.get(STATIC_CONFIG['api']['endpoints']['actions'])

27def do_action(self, sysargs: List[str]) -> SuccessTuple:
28    """
29    Execute a Meerschaum action remotely.
30    """
31    return self._rust_client().do_action(sysargs)

34async def do_action_async(
35    self,
36    sysargs: List[str],
37    callback_function: Callable[[str], None] = partial(print, end=''),
38) -> SuccessTuple:
39    """
40    Execute an action as a temporary remote job.
41    """
42    from meerschaum.connectors.api._jobs import _LoopCallback
43    loop = asyncio.get_running_loop()
44    client = self._rust_client()
45    return await loop.run_in_executor(
46        None,
47        lambda: client.do_action(sysargs, _LoopCallback(loop, callback_function)),
48    )

51def do_action_legacy(
52    self,
53    action: Optional[List[str]] = None,
54    sysargs: Optional[List[str]] = None,
55    debug: bool = False,
56    **kw
57) -> SuccessTuple:
58    """
59    NOTE: This method is deprecated.
60    Please use `do_action()` or `do_action_async()`.
61
62    Execute a Meerschaum action remotely.
63
64    If `sysargs` are provided, parse those instead.
65    Otherwise infer everything from keyword arguments.
66
67    Examples
68    --------
69    >>> conn = mrsm.get_connector('api:main')
70    >>> conn.do_action(['show', 'pipes'])
71    (True, "Success")
72    >>> conn.do_action(['show', 'arguments'], name='test')
73    (True, "Success")
74    """
75    if action is None:
76        action = []
77
78    if sysargs is not None and action and action[0] == '':
79        from meerschaum._internal.arguments import parse_arguments
80        json_dict = parse_arguments(sysargs)
81        action = list(json_dict.pop('action', None) or [])
82        kw = json_dict
83    else:
84        action = list(action)
85        if debug:
86            kw['debug'] = debug
87
88    return self._rust_client().do_action_legacy(action, kw)

15def get_mrsm_version(self, **kw) -> Optional[str]:
16    """
17    Return the Meerschaum version of the API instance.
18    """
19    return self._rust_client().get_mrsm_version()

22def get_chaining_status(self, **kw) -> Optional[bool]:
23    """
24    Fetch the chaining status of the API instance.
25    """
26    return self._rust_client().get_chaining_status()

5def get_engine_metrics(self, **kwargs):
6    """Read bounded sync and Pipeline measurements visible to this credential."""
7    from meerschaum.core.Pipeline._metrics import get_engine_metrics as read
8    return read(instance=self, **kwargs)

11def initialize_engine_metrics(self):
12    """Initialize metrics storage on the served instance (full administrator only)."""
13    from meerschaum.core.Pipeline._metrics import initialize_engine_metrics as initialize
14    return initialize(instance=self)

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

42def register_pipe(
43    self,
44    pipe: mrsm.Pipe,
45    debug: bool = False,
46) -> SuccessTuple:
47    """
48    Submit a POST to the API to register a new Pipe object.
49    Returns a tuple of (success_bool, response_dict).
50    """
51    return self._rust_client().register_pipe(pipe)

 67def fetch_pipes_keys(
 68    self,
 69    connector_keys: Optional[List[str]] = None,
 70    metric_keys: Optional[List[str]] = None,
 71    location_keys: Optional[List[str]] = None,
 72    tags: Optional[List[str]] = None,
 73    params: Optional[Dict[str, Any]] = None,
 74    debug: bool = False
 75) -> Union[
 76    Dict[Union[int, str], Tuple[str, str, Union[str, None], Dict[str, Any]]],
 77    List[Tuple[str, str, Union[str, None]]],
 78]:
 79    """
 80    Fetch registered Pipes' keys from the API.
 81
 82    Parameters
 83    ----------
 84    connector_keys: Optional[List[str]], default None
 85        The connector keys for the query.
 86
 87    metric_keys: Optional[List[str]], default None
 88        The metric keys for the query.
 89
 90    location_keys: Optional[List[str]], default None
 91        The location keys for the query.
 92
 93    tags: Optional[List[str]], default None
 94        A list of tags for the query.
 95
 96    params: Optional[Dict[str, Any]], default None
 97        A parameters dictionary for filtering against the `pipes` table
 98        (e.g. `{'connector_keys': 'plugin:foo'}`).
 99        Not recommeded to be used.
100
101    Returns
102    -------
103    A dictionary mapping pipe IDs to key tuples, or a list of key tuples for older servers.
104    """
105    return self._rust_client().fetch_pipes_keys(
106        connector_keys,
107        metric_keys,
108        location_keys,
109        tags,
110        params,
111    )

54def edit_pipe(
55    self,
56    pipe: mrsm.Pipe,
57    patch: bool = False,
58    debug: bool = False,
59) -> SuccessTuple:
60    """
61    Submit a PATCH to the API to edit an existing Pipe object.
62    Returns a tuple of (success_bool, response_dict).
63    """
64    return self._rust_client().edit_pipe(pipe, patch=patch)

114def sync_pipe(
115    self,
116    pipe: mrsm.Pipe,
117    df: Optional[Union['pd.DataFrame', Dict[Any, Any], str]] = None,
118    chunksize: Optional[int] = -1,
119    debug: bool = False,
120    **kw: Any
121) -> SuccessTuple:
122    """
123    Sync a DataFrame (or documents, or a dictionary of lists) into a Pipe.
124    Keyword arguments are sent to the API as query parameters.
125    """
126    if isinstance(df, str):
127        df = json.loads(df)
128    return self._rust_client().sync_pipe(pipe, df, chunksize=chunksize, **kw)

141def delete_pipe(
142    self,
143    pipe: Optional[mrsm.Pipe] = None,
144    debug: bool = False,
145) -> SuccessTuple:
146    """Delete a Pipe and drop its table."""
147    if pipe is None:
148        raise ValueError("Pipe cannot be None.")
149    return self._rust_client().delete_pipe(pipe)

131def delete_pipe_cache(
132    self,
133    pipe: mrsm.Pipe,
134    debug: bool = False,
135    **kw: Any
136) -> SuccessTuple:
137    """Invalidate the server-side cache for a pipe."""
138    return self._rust_client().delete_pipe_cache(pipe)

152def get_pipe_data(
153    self,
154    pipe: mrsm.Pipe,
155    select_columns: Optional[List[str]] = None,
156    omit_columns: Optional[List[str]] = None,
157    begin: Union[str, datetime, int, None] = None,
158    end: Union[str, datetime, int, None] = None,
159    params: Optional[Dict[str, Any]] = None,
160    as_chunks: bool = False,
161    debug: bool = False,
162    as_polars: Optional[bool] = None,
163    as_pandas: Optional[bool] = None,
164    **kw: Any
165) -> Union['pd.DataFrame', None]:
166    """
167    Fetch data from the API, as a DataFrame of the `meerschaum:dataframe` library
168    (`as_polars=True` / `as_pandas=True` choose one).
169    """
170    from meerschaum.utils.dataframe import frame_from_arrow, frame_library_of
171    df = self._rust_client().get_pipe_data(
172        pipe,
173        select_columns=select_columns,
174        omit_columns=omit_columns,
175        begin=begin,
176        end=end,
177        params=params,
178        as_chunks=as_chunks,
179    )
180    if frame_library_of(df) is None:
181        return df
182    return frame_from_arrow(df, as_polars=as_polars, as_pandas=as_pandas)

185def get_pipe_id(
186    self,
187    pipe: mrsm.Pipe,
188    debug: bool = False,
189) -> Union[int, str, None]:
190    """Get a Pipe's ID from the API."""
191    return self._rust_client().get_pipe_id(pipe)

194def get_pipe_attributes(
195    self,
196    pipe: mrsm.Pipe,
197    debug: bool = False,
198) -> Dict[str, Any]:
199    """
200    Get a Pipe's attributes from the API.
201
202    Parameters
203    ----------
204    pipe: meerschaum.Pipe
205        The pipe whose attributes we are fetching.
206
207    Returns
208    -------
209    A dictionary of a pipe's attributes.
210    If the pipe does not exist, return an empty dictionary.
211    """
212    return self._rust_client().get_pipe_attributes(pipe)

215def get_sync_time(
216    self,
217    pipe: mrsm.Pipe,
218    params: Optional[Dict[str, Any]] = None,
219    newest: bool = True,
220    debug: bool = False,
221) -> Union[datetime, int, None]:
222    """
223    Get a Pipe's most recent datetime value from the API.
224
225    Parameters
226    ----------
227    pipe: meerschaum.Pipe
228        The pipe to select from.
229
230    params: Optional[Dict[str, Any]], default None
231        Optional params dictionary to build the WHERE clause.
232
233    newest: bool, default True
234        If `True`, get the most recent datetime (honoring `params`).
235        If `False`, get the oldest datetime (ASC instead of DESC).
236
237    Returns
238    -------
239    The most recent (or oldest if `newest` is `False`) datetime of a pipe,
240    rounded down to the closest minute.
241    """
242    return self._rust_client().get_sync_time(pipe, params=params, newest=newest)

245def pipe_exists(
246    self,
247    pipe: mrsm.Pipe,
248    debug: bool = False
249) -> bool:
250    """
251    Check the API to see if a Pipe exists.
252
253    Parameters
254    ----------
255    pipe: 'meerschaum.Pipe'
256        The pipe which were are querying.
257
258    Returns
259    -------
260    A bool indicating whether a pipe's underlying table exists.
261    """
262    return self._rust_client().pipe_exists(pipe)

265def create_metadata(
266    self,
267    debug: bool = False
268) -> bool:
269    """
270    Create metadata tables.
271
272    Returns
273    -------
274    A bool indicating success.
275    """
276    return self._rust_client().create_metadata()

279def get_pipe_rowcount(
280    self,
281    pipe: mrsm.Pipe,
282    begin: Union[str, datetime, int, None] = None,
283    end: Union[str, datetime, int, None] = None,
284    params: Optional[Dict[str, Any]] = None,
285    remote: bool = False,
286    debug: bool = False,
287) -> int:
288    """
289    Get a pipe's row count from the API.
290
291    Parameters
292    ----------
293    pipe: 'meerschaum.Pipe':
294        The pipe whose row count we are counting.
295
296    begin: Union[str, datetime, int, None], default None
297        If provided, bound the count by this datetime.
298
299    end: Union[str, datetime, int, None], default None
300        If provided, bound the count by this datetime.
301
302    params: Optional[Dict[str, Any]], default None
303        If provided, bound the count by these parameters.
304
305    remote: bool, default False
306        If `True`, return the rowcount for the fetch definition.
307
308    Returns
309    -------
310    The number of rows in the pipe's table, bound the given parameters.
311    If the table does not exist, return 0.
312    """
313    return self._rust_client().get_pipe_rowcount(
314        pipe,
315        begin=begin,
316        end=end,
317        params=params,
318        remote=remote,
319    )

322def drop_pipe(
323    self,
324    pipe: mrsm.Pipe,
325    debug: bool = False
326) -> SuccessTuple:
327    """
328    Drop a pipe's table but maintain its registration.
329
330    Parameters
331    ----------
332    pipe: meerschaum.Pipe:
333        The pipe to be dropped.
334
335    Returns
336    -------
337    A success tuple (bool, str).
338    """
339    if pipe is None:
340        raise ValueError("Pipe cannot be None.")
341    return self._rust_client().drop_pipe(pipe)

344def clear_pipe(
345    self,
346    pipe: mrsm.Pipe,
347    begin: Union[str, datetime, int, None] = None,
348    end: Union[str, datetime, int, None] = None,
349    params: Optional[Dict[str, Any]] = None,
350    debug: bool = False,
351    **kw
352) -> SuccessTuple:
353    """
354    Delete rows in a pipe's table.
355
356    Parameters
357    ----------
358    pipe: meerschaum.Pipe
359        The pipe with rows to be deleted.
360
361    begin: Union[str, datetime, int, None], default None
362        Delete rows at or after this datetime.
363
364    end: Union[str, datetime, int, None], default None
365        Delete rows before this datetime.
366
367    params: Optional[Dict[str, Any]], default None
368        Delete rows matching these parameters.
369
370    Returns
371    -------
372    A success tuple.
373    """
374    return self._rust_client().clear_pipe(pipe, begin=begin, end=end, params=params)

377def get_pipe_columns_types(
378    self,
379    pipe: mrsm.Pipe,
380    debug: bool = False,
381) -> Union[Dict[str, str], None]:
382    """
383    Fetch the columns and types of the pipe's table.
384
385    Parameters
386    ----------
387    pipe: meerschaum.Pipe
388        The pipe whose columns to be queried.
389
390    Returns
391    -------
392    A dictionary mapping column names to their database types, or `None` on failure.
393    """
394    return self._rust_client().get_pipe_columns_types(pipe)