diff --git a/README.md b/README.md index 8ecdd36..eb0c563 100644 --- a/README.md +++ b/README.md @@ -72,8 +72,8 @@ Implement following REST endpoints to manage requests with Sinequa API. - [x] search.labels - [x] serach.savedQueries - [x] search.suggest -- [ ] search.custom -- [ ] suggestField +- [x] search.custom +- [x] suggestField **Indexing Endpoints** - [ ] indexing.collection diff --git a/pynequa/core.py b/pynequa/core.py index 47b6fa1..be5a92c 100644 --- a/pynequa/core.py +++ b/pynequa/core.py @@ -2,7 +2,12 @@ from typing import Optional, List, Dict, Literal from pynequa.api import API -from pynequa.models import QueryParams, TreeParams, AlertParams +from pynequa.models import ( + QueryParams, + TreeParams, + AlertParams, + CustomSearchParams +) class Sinequa(API): @@ -504,19 +509,41 @@ def search_suggest(self, app: str, profile: str, suggestion_query: str, payload["kinds"] = kinds return self.post(endpoint=endpoint, payload=payload) - def search_custom(self): + def search_custom(self, params: CustomSearchParams) -> Dict: ''' Define and run customized search on indexes. Warning: Requires Admin privileges. ''' endpoint = "search.custom" - pass + payload = params.generate_payload() + return self.post(endpoint=endpoint, payload=payload) - def suggest_field(self): + def suggest_field(self, profile: str, action: str = Literal["suggests", "defaultFields"], + fields: List[str] = [], + text: str = "") -> Dict: ''' + suggest_field provides a suggestion for a fielded search. + + Args: + profile(str): name of profile + action(str): action to be performed + fields(List[str]): list of fields for suggestions (required with action=suggests) + text(str): text to be searched for suggestions (required with action=suggests) + Returns: + Dict: response for suggest field ''' endpoint = "suggestField" - pass + payload = { + "profile": profile, + "app": self.app_name, + "action": action, + } + + if action == "suggests": + payload["fields"] = fields + payload["text"] = text + + return self.post(endpoint=endpoint, payload=payload) def engine_sql(self, sql: str, max_rows: int = 1000) -> Dict: ''' diff --git a/pynequa/models.py b/pynequa/models.py index 19e2ab5..d2438e4 100644 --- a/pynequa/models.py +++ b/pynequa/models.py @@ -359,3 +359,126 @@ def generate_payload(self, **kwargs) -> Dict: if self.debug: logger.debug(payload) return payload + + +@dataclass +class CustomSearchParams(AbstractParams): + + @dataclass + class ConditionParam: + column: str + type: str + value: Optional[str] = None + cache: bool = False + + def generate_payload(self) -> Dict: + payload = { + "column": self.column, + "type": self.type, + "cache": self.cache, + } + if self.value is not None: + payload["value"] = self.value + return payload + + @dataclass + class SelectionParam: + type: str + column: Optional[str] = None + with_null: bool = False + values: Optional[str] = None + + def generate_payload(self) -> Dict: + payload = { + "type": self.type, + "withNull": self.with_null + } + if self.column is not None: + payload["column"] = self.column + if self.values is not None: + payload["values"] = self.values + return payload + + @dataclass + class NavigationBoxParam: + column: str + name: str + display: Optional[str] = None + mask: Optional[Literal["YYYY-MM-DD", "YYYY-MM", + "YYYY", "MM", "HH"]] = None + order: Optional[Literal["freqdesc", "freqasc", "keydesc", + "keyasc", "valuedesc", "valueasc", + "none"]] = None + type: Optional[Literal["list", "tree", "concepts", "tag cloud", + "merge list", "breadcrumb", "advanced filters", + "recent queries", "audit"]] = None + want_nulls: bool = False + + def generate_payload(self) -> Dict: + payload = { + "column": self.column, + "name": self.name, + "wantnulls": self.want_nulls + } + + if self.display is not None: + payload["display"] = self.display + if self.mask is not None: + payload["mask"] = self.mask + if self.order is not None: + payload["order"] = self.order + if self.type is not None: + payload["type"] = self.type + return payload + + index_list: List[str] + text: str = None + rows_as_objects: bool = False + conditions: List[ConditionParam] = None + selections: List[SelectionParam] = None + facets: List[NavigationBoxParam] = None + columns: List[str] = None + skip_count: int = 20 + skip_from: int = None + index_list: List[str] = None + engine_list: List[str] = None + search_parameters: str = None + + def _prepare_payload(self) -> Dict: + payload = { + "indexList": self.index_list, + "rowsAsObjects": self.rows_as_objects, + "skipCount": self.skip_count, + } + + if self.text is not None: + payload["text"] = self.text + + if self.columns is not None: + payload["columns"] = self.columns + + if self.skip_from is not None: + payload["skipFrom"] = self.skip_from + + if self.index_list is not None: + payload["indexList"] = self.index_list + + if self.engine_list is not None: + payload["engineList"] = self.engine_list + + if self.conditions is not None: + payload["conditions"] = [item.generate_payload() + for item in self.conditions] + + if self.selections is not None: + payload["selections"] = [item.generate_payload() + for item in self.selections] + + if self.facets is not None: + payload["facets"] = [item.generate_payload() + for item in self.facets] + + return payload + + def generate_payload(self, **kwargs) -> Dict: + return self._prepare_payload() diff --git a/tests/test_models.py b/tests/test_models.py index 1f1ee41..2e45565 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -1,4 +1,4 @@ -from pynequa.models import QueryParams, AdvancedParams +from pynequa.models import QueryParams, AdvancedParams, AlertParams import unittest import logging import json @@ -86,5 +86,34 @@ def test_query_params_with_advanced_params(self): assert payload["advanced"] == expected_payload +class TestAlertParams(unittest.TestCase): + + def test_alert_params_payload(self): + """ + Test if alert params payload is correctly + generated or not. + """ + ap = AlertParams( + name="TestAlert", + description="This is a test alert", + frequency="hourly", + combine_with_other_alerts=True, + respect_tab_selection=True, + ) + + generated_payload = ap.generate_payload() + + expected_payload = { + "name": "TestAlert", + "description": "This is a test alert", + "frequency": "hourly", + "active": False, + "combineWithOtherAlerts": True, + "respectTabSelection": True + } + + assert generated_payload == expected_payload + + if __name__ == '__main__': unittest.main()