From 80b4d4cfee7f82090cc821fd613a6a72e3d8f9f8 Mon Sep 17 00:00:00 2001 From: lastorel Date: Mon, 8 Aug 2022 02:06:22 +0300 Subject: [PATCH] #16 Add search engine --- CHANGELOG.md | 1 + README.md | 14 ++++++++++++++ pytion/api.py | 45 +++++++++++++++++++++++++++++++++++++-------- pytion/query.py | 24 +++++++++++++++++++----- tests/test_api.py | 22 +++++++++++++++++++++- 5 files changed, 92 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f022fc..ef103e7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ - [#27](https://github.com/lastorel/pytion/issues/27): updates for `relation` type `PropertyValue` - [#16](https://github.com/lastorel/pytion/issues/17): tests of Property model - [#28](https://github.com/lastorel/pytion/issues/28): Add whoami method +- [#16](https://github.com/lastorel/pytion/issues/16): Add search engine ### Breaking changes diff --git a/README.md b/README.md index 2801c85..519a741 100644 --- a/README.md +++ b/README.md @@ -133,6 +133,19 @@ new_page = page.page_update(title="new page name 2") # new_page.obj is equal page.obj except title and last_edited properties ``` +### Search + +There is a search example: +```python +no = Notion(token) + +r = no.search("updating", object_type="page") +print(r.obj) +# output: +# Page for updating +# Page to updating databases +``` + ### pytion.api.Element @@ -230,6 +243,7 @@ There are also useful **internal** classes: - You can create object `LinkTo.create()` and use it in many places and methods - use `LinkTo(from_object=my_page1)` to quickly create a link to any existing object of pytion.models - `link` property of `LinkTo` returns expanded URL +- `ElementArray` is found while using `.search()` endpoint. It's a parent of `PageArray` > And every model has a `.get()` method that returns API friendly JSON. diff --git a/pytion/api.py b/pytion/api.py index 09469fe..8e54b0c 100644 --- a/pytion/api.py +++ b/pytion/api.py @@ -10,7 +10,7 @@ from pytion.models import ElementArray, User -Models = Union[Database, Page, Block, BlockArray, PropertyValue, PageArray] +Models = Union[Database, Page, Block, BlockArray, PropertyValue, PageArray, ElementArray] logger = logging.getLogger(__name__) @@ -26,15 +26,44 @@ def __init__(self, token: Optional[str] = None, version: Optional[str] = None): self.session = Request(api=self, token=token) logger.debug(f"API object created. Version {envs.NOTION_VERSION}") + def search( + self, query: Optional[str] = None, limit: int = 0, + object_type: Optional[str] = None, sort_last_edited_time: Optional[str] = None + ) -> Optional[Element]: + """ + Searches all original pages, databases, and child pages/databases that are shared with the integration. + It will not return linked databases, since these duplicate their source databases. (c) + + + :param query: search by page title + :param limit: 0 < int < 100 - max number of items to be returned (0 = return all) + :param object_type: filter by type: 'page' or 'database' + :param sort_last_edited_time: sorting 'ascending' or 'descending' + :return: + + `r = no.search("pytion", 10, sort_last_edited_time="ascending")` + `print(r.obj)` + """ + data = {"query": query} if query else None + filter_ = Filter(raw={"property": "object", "value": object_type}) if object_type else None + if sort_last_edited_time: + sort_last_edited_time = Sort(property_name="last_edited_time", direction=sort_last_edited_time) + result = self.session.method( + "post", "search", sort=sort_last_edited_time, filter_=filter_, limit=limit, data=data + ) + if "results" in result and isinstance(result["results"], list): + data = ElementArray(result["results"]) + for item in data: + if isinstance(item, Page): + self.pages.get_page_properties(title_only=True, obj=item) + return Element(api=self, name="search", obj=data) + else: + logger.warning("Results list is not found") + return None + def __len__(self): return 1 - # def __getstate__(self): - # return {"api": self.session} - # - # def __setstate__(self, d): - # self.__dict__.update(d) - def __repr__(self): return "NotionAPI" @@ -59,7 +88,7 @@ def __init__(self, api: Notion, name: str, obj: Optional[Models] = None): def get(self, id_: str, _after_path: str = None, limit: int = 0) -> Element: """ Get Element by ID. - .query.RequestError exception if not found + .exceptions.ObjectNotFound exception if not found :return: `Element.obj` may be `Page`, `Database`, `Block` diff --git a/pytion/query.py b/pytion/query.py index 1ce85ab..19182a3 100644 --- a/pytion/query.py +++ b/pytion/query.py @@ -119,17 +119,25 @@ class Sort(object): directions = ["ascending", "descending"] def __init__(self, property_name: str, direction: str = "ascending"): + """ + Sort object is used while querying database or search query: + - self.sort object is used in search query (only single item supported by API) + - self.sorts can contain multiple criteria and is used in database query + """ if direction not in self.directions: raise ValueError(f"Allowed types {self.directions} ({direction} is provided)") if property_name in ("created_time", "last_edited_time"): - self.sorts = [{"timestamp": property_name, "direction": direction}] + self.sort = {"timestamp": property_name, "direction": direction} + self.sorts = [self.sort] else: - self.sorts = [{"property": property_name, "direction": direction}] + self.sort = {"property": property_name, "direction": direction} + self.sorts = [self.sort] def add(self, property_name: str, direction: str): if direction not in self.directions: raise ValueError(f"Allowed types {self.directions} ({direction} is provided)") - self.sorts.append({"property": property_name, "direction": direction}) + self.sort = {"property": property_name, "direction": direction} + self.sorts.append(self.sort) def __repr__(self): r = [e.values() for e in self.sorts] @@ -166,8 +174,9 @@ def __init__( self.result = self.method(method, path, id_, data, after_path, limit, filter_, sorts) def method( - self, method, path, id_="", data=None, after_path=None, - limit=0, filter_: Optional[Filter] = None, sorts: Optional[Sort] = None, pagination_loop: bool = False, + self, method: str, path: str, id_: str = "", data: Optional[Dict] = None, + after_path: Optional[str] = None, limit: int = 0, filter_: Optional[Filter] = None, + sorts: Optional[Sort] = None, pagination_loop: bool = False, sort: Optional[Sort] = None, ): if filter_: if data: @@ -179,6 +188,11 @@ def method( data["sorts"] = sorts.sorts else: data = {"sorts": sorts.sorts} + if sort: # specific attr in 'search' query. strange + if data: + data["sort"] = sort.sort + else: + data = {"sort": sort.sort} url = self.base + path + "/" + id_ if limit and method == "get": if after_path: diff --git a/tests/test_api.py b/tests/test_api.py index 1191093..3d99dbb 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -2,7 +2,7 @@ import pytest -from pytion.models import Page, Block, Database, User, RichTextArray +from pytion.models import Page, Block, Database, User, RichTextArray, ElementArray from pytion.models import BlockArray, PropertyValue, PageArray, LinkTo, Property from pytion import InvalidRequestURL, ObjectNotFound, ValidationError @@ -12,6 +12,26 @@ def test_notion(no): assert no.session.base == "https://api.notion.com/v1/" +def test_search__empty(no): + r = no.search("123412341234") + assert isinstance(r.obj, ElementArray) + assert len(r.obj) == 0 + + +def test_search__type_and_limit(no): + r = no.search(object_type="database", limit=4) + assert isinstance(r.obj, ElementArray) + assert len(r.obj) == 4 + assert all(isinstance(item, Database) for item in r.obj) + + +def test_search__query(no): + r = no.search("tests") + assert isinstance(r.obj, ElementArray) + assert len(r.obj) >= 1 + assert str(r.obj[0]) == "Pytion Tests" + + class TestElement: def test_get__page(self, root_page): page = root_page