diff --git a/README.md b/README.md index 5f6aedd..8ecdd36 100644 --- a/README.md +++ b/README.md @@ -67,11 +67,11 @@ Implement following REST endpoints to manage requests with Sinequa API. - [x] search.ratings - [x] search.profile.subtree - [x] engine.sql -- [ ] search.alerts -- [ ] search.baskets -- [ ] search.labels -- [ ] serach.savedQueries -- [ ] search.suggest +- [x] search.alerts +- [x] search.baskets +- [x] search.labels +- [x] serach.savedQueries +- [x] search.suggest - [ ] search.custom - [ ] suggestField diff --git a/pynequa/core.py b/pynequa/core.py index 09b56c3..47b6fa1 100644 --- a/pynequa/core.py +++ b/pynequa/core.py @@ -1,8 +1,8 @@ from __future__ import annotations -from typing import Optional, List, Dict +from typing import Optional, List, Dict, Literal from pynequa.api import API -from pynequa.models import QueryParams, TreeParams +from pynequa.models import QueryParams, TreeParams, AlertParams class Sinequa(API): @@ -361,38 +361,153 @@ def search_profile_subtree(self, profile: str, query_params: QueryParams, } return self.post(endpoint=endpoint, payload=payload) - def search_alerts(self): + def search_alerts(self, action: Literal["list", "create", "read", "update", "delete"], + alert: AlertParams, + name: Optional[str] = "", old_name: Optional[str] = ""): ''' + This method makes it possible to manage alerts in search.alerts endpoint. + Args: + action(str): action name, options: [list, create, read, update, delete] + alert_params(AlertParams): parameters for alert configuration + name(str): name of alert to be read [OPTIONAL] + old_name(str): name of alert before you change it [OPTIONAL] + Returns: + Dict: returns a JSON response with listing alerts ''' endpoint = "search.alerts" - pass + payload = { + "action": action, + "alert": alert.generate_payload(), + "name": name, + "oldName": old_name + } + return self.post(endpoint=endpoint, payload=payload) - def search_baskets(self): + def search_baskets(self, action: Literal["list", "get", "create", "delete", + "update", "adddocs", "deletedocs"], + name: Optional[str] = None, + new_name: Optional[str] = None, + id: Optional[str] = None) -> Dict: ''' + This method makes it possible to manage baskets. + Args: + action(str): action name, options: [list, get, create, delete, update, adddocs, deletedocs] + name(str): sets the name of basket [OPTIONAL] + new_name(str): new name of basket [OPTIONAL] + id(str): list ids of documents to be added or deleted from basket [OPTIONAL] + Returns: + Dict: returns a JSON response ''' endpoint = "search.baskets" - pass + payload = { + "action": action, + } + if name is not None: + payload["name"] = name + if new_name is not None: + payload["newName"] = new_name + if id is not None: + payload["id"] = id + + return self.post(endpoint=endpoint, payload=payload) - def search_labels(self): + def search_labels(self, action: Literal["list", "add", "remove", "rename", + "delete", "bulkAdd", "bulkRemove"], + q: Optional[str] = None, + id: Optional[str] = None, + labels: Optional[str] = None, + new_label: Optional[str] = None, + public: Optional[bool] = False, + query: Optional[str] = None) -> Dict: ''' + This method makes it possible to manage labels. + + Args: + action(str): action name, options: [list, add, remove, rename, delete, bulkAdd, bulkRemove] + q(str): label prefix + id(str): document id(s) + labels(str): name of labels to be added, removed etc + new_label(str): new name of label + public(bool): whether labels are public or not + query(str): query string + Returns: + Dict: returns a JSON response ''' endpoint = "search.labels" - pass + payload = { + "action": action, + "public": public + } + if q is not None: + payload["q"] = q + if id is not None: + payload["id"] = id + if labels is not None: + payload["labels"] = labels + if new_label is not None: + payload["newLabel"] = new_label + if public is not None: + payload["public"] = public + if query is not None: + payload["query"] = query + return self.post(endpoint=endpoint, payload=payload) - def search_saved_queries(self): + def search_saved_queries(self, action: Literal["list", "get", "set", "delete"], + name: Optional[str] = None, + saved_query: Optional[Dict] = None) -> Dict: ''' + Manage saved queries. + + Args: + action(str): action name, options: [list, get, set, delete] + name(str): name of saved query + saved_query(Dict): parameters of new saved query to be created. ''' endpoint = "search.savedQueries" - pass + payload = { + "action": action, + } - def search_suggest(self): + if name is not None: + payload["name"] = name + if saved_query is not None: + payload["savedQuery"] = saved_query + return self.post(endpoint=endpoint, payload=payload) + + def search_suggest(self, app: str, profile: str, suggestion_query: str, + text: str, filter: Optional[Dict] = None, + kinds: Optional[List[str]] = None, + show_source: Optional[bool] = False) -> Dict: ''' + Manage auto complete suggestions + + Args: + app(str): name of SBA application + profile(str): name of profile + suggestion_query(str): name of suggestion query to be used + text(str): text to be searche for + filter(Dict): key/value pairs to filter + kinds(List[str]): kinds of suggestions + show_source(bool): returns name of index column when True ''' endpoint = "search.suggest" - pass + payload = { + "app": app, + "profile": profile, + "suggestionQuery": suggestion_query, + "text": text, + "showSource": show_source, + } + if filter is not None: + payload["filter"] = filter + if kinds is not None: + payload["kinds"] = kinds + return self.post(endpoint=endpoint, payload=payload) def search_custom(self): ''' + Define and run customized search on indexes. + Warning: Requires Admin privileges. ''' endpoint = "search.custom" pass diff --git a/pynequa/models.py b/pynequa/models.py index 32d8170..19e2ab5 100644 --- a/pynequa/models.py +++ b/pynequa/models.py @@ -1,4 +1,4 @@ -from typing import Dict, List, Optional, Union +from typing import Dict, List, Optional, Union, Literal from abc import abstractmethod, ABC from dataclasses import dataclass, field from loguru import logger @@ -293,3 +293,69 @@ def generate_payload(self, **kwargs) -> Dict: logger.debug(payload) return payload + + +@dataclass +class AlertParams(AbstractParams): + """ + AlertParams define the set of parameters available + to configure an alert or update it. + """ + name: str + description: Optional[str] = None + profile: Optional[str] = None + timezone: Optional[str] = None + frequency: Optional[Literal["daily", "hourly", + "immediate", "weekly", "monthly"]] = "daily" + days: Optional[str] = None + alert_from: Optional[str] = None + alert_to: Optional[str] = None + times: Optional[str] = None # format: HH:mm + active: bool = False + combine_with_other_alerts: bool = False + respect_tab_selection: bool = False + debug: bool = False + + def _prepare_alert_params_payload(self) -> Dict: + params = { + "name": self.name, + "active": self.active, + "combineWithOtherAlerts": self.combine_with_other_alerts, + "respectTabSelection": self.respect_tab_selection + } + + if self.description is not None: + params["description"] = self.description + + if self.profile is not None: + params["profile"] = self.profile + + if self.timezone is not None: + params["timezone"] = self.timezone + + if self.frequency is not None: + params["frequency"] = self.frequency + + if self.days is not None: + params["days"] = self.days + + if self.alert_from is not None: + params["from"] = self.alert_from + + if self.alert_to is not None: + params["to"] = self.alert_to + + if self.times is not None: + params["times"] = self.times + + return params + + def generate_payload(self, **kwargs) -> Dict: + """ + This method generates payload for + AlertParams. + """ + payload = self._prepare_alert_params_payload() + if self.debug: + logger.debug(payload) + return payload