Add database support + refactor as necessary

Author: floscha

notion/client.py

@@ -1,8 +1,11 @@
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional, Union
 
 import requests
 
-from notion.model import Page
+from notion.model.common.utils import UUIDv4
+from notion.model.databases.database import Database
+from notion.model.filters import Filter
+from notion.model.page import Page
 
 API_BASE_URL = "https://api.notion.com/v1/"
 API_VERSION = "2022-02-22"
@@ -41,7 +44,7 @@ def _paginate(
         entity: str,
         payload: Optional[Dict[str, Any]] = None,
         limit: Optional[int] = None,
-    ) -> list:
+    ) -> List[dict]:
         if payload is None:
             payload = {}
 
@@ -59,21 +62,65 @@ def _paginate(
         return results[:limit]
 
     # ---------------------------------------------------------------------------
-    # Pages
+    # Databases
     # ---------------------------------------------------------------------------
 
-    def get_pages(self, database_id, filter_: Optional[dict] = None):
-        return self._paginate("post", f"databases/{database_id}/query", filter_)
+    def get_database(self, database_id) -> Database:
+        "Get a single Notion database by its ID."
+        data = self._make_request("get", f"databases/{database_id}")
+        return Database.from_json(data).with_client(self)
+
+    def query_database(
+        self,
+        database_id,
+        filter_: Optional[Union[Filter, dict]] = None,
+        sort: Optional[dict] = None,
+    ) -> List[Page]:
+        "Query a Notion database for pages given some filter(s)."
+        filter_ = {
+            "filter": filter_.to_json() if isinstance(filter_, Filter) else filter_
+        }
+        data = self._paginate(
+            "post", f"databases/{database_id}/query", {**filter_, **(sort or {})}
+        )
+        return [Page.from_json(page_data).with_client(self) for page_data in data]
+
+    def create_database(self, database: Database, parent_id: Optional[UUIDv4] = None):
+        "Create a new Notion database."
+        if parent_id:
+            database._data["parent"] = {"type": "page_id", "page_id": parent_id}
+        response = self._make_request("post", "databases", database._data)
+        database._data = response
+        database._client = self
+
+    def update_database(self, database_id, payload: dict) -> dict:
+        "Update properties of an existing Notion database."
+        return self._make_request("patch", f"databases/{database_id}", payload)
+
+    def delete_database(self, page_id):
+        """Deletes the Notion Page with the given ID.
+
+        The Notion API does not offer a DELETE method but insteads works by setting the `archived` field.
+        """
+        return self.update_database(page_id, {"archived": True})
+
+    # ---------------------------------------------------------------------------
+    # Pages
+    # ---------------------------------------------------------------------------
 
     def get_page(self, page_id):
+        "Get a single Notion page by its ID."
         data = self._make_request("get", f"pages/{page_id}")
         return Page(client=self, data=data)
 
-    def create_page(self, page) -> Page:
-        response = self._make_request("post", "pages", page)
+    def create_page(self, page: Union[Page, dict]) -> Page:
+        "Create a new Notion page."
+        page_data = page.to_json() if isinstance(page, Page) else page
+        response = self._make_request("post", "pages", page_data)
         return response
 
     def update_page(self, page_id, payload: dict):
+        "Update properties of an existing Notion page."
         return self._make_request("patch", f"pages/{page_id}", payload)
 
     def delete_page(self, page_id):
@@ -88,12 +135,15 @@ def delete_page(self, page_id):
     # ---------------------------------------------------------------------------
 
     def update_block(self, block_id, payload: dict):
+        "Update properties of an existing Notion page."
         return self._make_request("patch", f"blocks/{block_id}", payload)
 
     def retrieve_block_children(self, block_id: str, limit: Optional[int] = None):
+        "Retrieve children of a given block."
         return self._make_request("get", f"blocks/{block_id}/children")
 
     def append_block_children(self, block_id: str, children: str):
+        "Append children blocks to an existing block"
         return self._make_request(
             "patch", f"blocks/{block_id}/children", {"children": children}
         )
@@ -105,6 +155,10 @@ def delete_block(self, block_id: str):
         """
         return self.update_block(block_id, {"archived": True})
 
+    # ---------------------------------------------------------------------------
+    # Search
+    # ---------------------------------------------------------------------------
+
     def search(
         self,
         query: str,
@@ -120,4 +174,4 @@ def search(
             payload["filter"] = filter
 
         results = self._paginate("post", "search", payload, limit)
-        return [Page(data=p, client=self) for p in results]
+        return [Page.from_json(page_data).with_client(self) for page_data in results]

notion/model/__init__.py

@@ -1,2 +1,2 @@
-from notion.model.block import *
-from notion.model.page import *
+from notion.model.databases import Database
+from notion.model.page import Page

notion/model/block.py

@@ -1,7 +1,7 @@
 from typing import List, Optional, Union
 
-from notion.model.common import NotionObjectBase, UUIDv4
-
+from notion.model.common.notion_object_base import NotionObjectBase
+from notion.model.common.utils import UUIDv4
 
 # ---------------------------------------------------------------------------
 # Base Class
@@ -20,6 +20,14 @@ def has_children(self) -> bool:
     def delete(self):
         self._data = self._client.delete_block(self.id)
 
+    def to_json(self) -> dict:
+        res = self._data.copy()
+        if "children" in res[self.type]:
+            res[self.type]["children"] = [
+                child.to_json() for child in res[self.type]["children"]
+            ]
+        return res
+
 
 # ---------------------------------------------------------------------------
 # Utils
@@ -29,6 +37,7 @@ def delete(self):
 def type_name_from_object(object) -> str:
     type_name = {
         ChildPage: "child_page",
+        ChildDatabase: "child_database",
         Paragraph: "paragraph",
         HeadingOne: "heading_1",
         HeadingTwo: "heading_2",
@@ -67,6 +76,7 @@ def type_name_from_object(object) -> str:
 def block_class_from_type_name(type_name: str) -> Block:
     type_class = {
         "child_page": ChildPage,
+        "child_database": ChildDatabase,
         "paragraph": Paragraph,
         "heading_1": HeadingOne,
         "heading_2": HeadingTwo,
@@ -129,7 +139,7 @@ def append_children(self, children: Union[dict, List[dict]]) -> List[dict]:
             object_name = child._data["object"]
             if object_name == "block":
                 append_results = self._client.append_block_children(
-                    self.id, [child.to_dict()]
+                    self.id, [child.to_json()]
                 )
                 new_block = append_results["results"][0]
                 child._data = new_block
@@ -244,17 +254,11 @@ def url(self, new_url: str) -> str:
 # ---------------------------------------------------------------------------
 
 
-class ChildPage(Block):
-    """A page contained in another page.
-
-    From the Notion docs (https://developers.notion.com/docs/working-with-page-content#modeling-content-as-blocks):
-        When a child page appears inside another page, it's represented as a `child_page` block, which does not have children.
-        You should think of this as a reference to the page block.
-    """
+class Child(Block):
+    "A block contained in another page."
 
     @property
     def title(self) -> str:
-        # return self._data["child_page"]["title"]
         return self._data[self.type]["title"]
 
     @property
@@ -266,15 +270,36 @@ def parent(self):
         full_page = self._client.get_page(self.id)
         return full_page.parent
 
+
+class ChildPage(Child):
+    """A page contained in another page.
+
+    From the Notion docs (https://developers.notion.com/docs/working-with-page-content#modeling-content-as-blocks):
+        When a child page appears inside another page, it's represented as a `child_page` block, which does not have children.
+        You should think of this as a reference to the page block.
+    """
+
     def delete(self):
-        """Delete the ChildPage.
+        """Delete the `ChildPage` in Notion.
 
         Needs to be overwritten to use the `delete_page` endpoint instead of `delete_block`.
         """
         deletion_result = self._client.delete_page(self.id)
         self._data["archived"] = deletion_result["archived"]
 
 
+class ChildDatabase(Child):
+    "A database contained in another page."
+
+    def delete(self):
+        """Delete the `ChildDatabase` in Notion.
+
+        Needs to be overwritten to use the `delete_database` endpoint instead of `delete_block`.
+        """
+        deletion_result = self._client.delete_database(self.id)
+        self._data["archived"] = deletion_result["archived"]
+
+
 class RichText(Block, RichTextMixin):
     def __init__(self, text: str = None, data=None, client=None) -> None:
         if not data:

notion/model/common/__init__.py

@@ -1,5 +1,11 @@
 from .notion_object_base import NotionObjectBase
-from .types import UUIDv4
-from .utils import is_valid_notion_id, parse_notion_datetime
+from .rich_text import RichText
+from .utils import UUIDv4, is_valid_notion_id, parse_notion_datetime
 
-__all__ = ["is_valid_notion_id", "NotionObjectBase", "parse_notion_datetime", "UUIDv4"]
+__all__ = [
+    "is_valid_notion_id",
+    "NotionObjectBase",
+    "parse_notion_datetime",
+    "RichText",
+    "UUIDv4",
+]

notion/model/common/emoji.py

@@ -0,0 +1,6 @@
+class Emoji:
+    def __init__(self, emoji: str):
+        self.emoji = emoji
+
+    def to_json(self) -> dict:
+        return {"type": "emoji", "emoji": self.emoji}

notion/model/common/file.py

@@ -0,0 +1,6 @@
+class File:
+    def __init__(self, url: str):
+        self.url = url
+
+    def to_json(self) -> dict:
+        return {"type": "external", "external": {"url": self.url}}

notion/model/common/notion_object_base.py

@@ -1,5 +1,7 @@
 from datetime import datetime
+from typing import Union
 
+from notion.model.common.parent import Parent, ParentPage
 from notion.model.common.utils import parse_notion_datetime
 
 
@@ -8,6 +10,10 @@ def __init__(self, data=None, client=None):
         self._data = data
         self._client = client
 
+    def with_client(self, client) -> "NotionObjectBase":
+        self._client = client
+        return self
+
     @property
     def object(self) -> str:
         """Get the Notion object type of the page as a string.
@@ -20,6 +26,17 @@ def object(self) -> str:
     def id(self) -> str:
         return self._data["id"]
 
+    @property
+    def parent(self) -> Parent:
+        return Parent.from_json(self._data)
+
+    @parent.setter
+    def parent(self, new_parent: Union[ParentPage, str]):
+        if isinstance(new_parent, str):
+            new_parent = ParentPage(new_parent)
+
+        self._data["parent"] = new_parent.to_json()
+
     @property
     def created_time(self) -> datetime:
         return parse_notion_datetime(self._data["created_time"])
@@ -43,11 +60,3 @@ def last_edited_by(self) -> dict:
     @property
     def archived(self) -> bool:
         return self._data["archived"]
-
-    def to_dict(self) -> dict:
-        res = self._data.copy()
-        if "children" in res[self.type]:
-            res[self.type]["children"] = [
-                child.to_dict() for child in res[self.type]["children"]
-            ]
-        return res

notion/model/common/parent.py

@@ -0,0 +1,58 @@
+from typing import Union
+
+from notion.model.common.utils import UUIDv4
+
+
+class Parent:
+    """The parent Notion object of another given Notion object.
+
+    Docs: https://developers.notion.com/reference/parent-object
+    """
+
+    def __init__(self, type_: str, id_: Union[UUIDv4, str, bool]):
+        if isinstance(id_, str) and type_ in ("page_id", "database_id"):
+            id_ = UUIDv4(id_)
+        elif isinstance(id_, bool) and id_ and type_ == "workspace":
+            pass
+        else:
+            raise ValueError("Invalid combination of `id_` and `type_` arguments.")
+
+        self.type = type_
+        self.id = id_
+
+    def from_json(
+        data: dict,
+    ) -> Union["ParentWorkspace", "ParentPage", "ParentDatabase", None]:
+        if "parent" not in data:
+            return None
+
+        type_ = data["parent"]["type"]
+        id_ = data["parent"][type_]
+
+        if type_ == "workspace":
+            return ParentWorkspace()
+        elif type_ == "page_id":
+            return ParentPage(id_)
+        elif type_ == "database_id":
+            return ParentDatabase(id_)
+        else:
+            # FIXME: This can never be reached because id_=... line will not find key.
+            raise ValueError(f"Parent type {type_!r} is not supported.")
+
+    def to_json(self):
+        return {"type": self.type, self.type: self.id}
+
+
+class ParentWorkspace(Parent):
+    def __init__(self):
+        super().__init__("workspace", True)
+
+
+class ParentPage(Parent):
+    def __init__(self, id_):
+        super().__init__("page_id", id_)
+
+
+class ParentDatabase(Parent):
+    def __init__(self, id_):
+        super().__init__("database_id", id_)

notion/model/common/rich_text.py

@@ -0,0 +1,21 @@
+class RichText:
+    def __init__(self, text: str):
+        self.text = text
+
+    @staticmethod
+    def from_json(data: dict) -> "RichText":
+        return RichText(data)
+
+    def to_json(self) -> dict:
+        return {
+            "type": "text",
+            "text": {
+                "content": self.text,
+            },
+        }
+
+    def __str__(self) -> str:
+        return self.text
+
+    def __repr__(self) -> str:
+        return str(self)