Type Hint Responses [WIP]

notion_client/_async_api_endpoints.py

@@ -0,0 +1,308 @@
+# mypy: disable-error-code="no-any-return"
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
+
+from notion_client.helpers import pick
+from notion_client.types import BotUserObjectResponse, UserObjectResponse
+
+if TYPE_CHECKING:
+    from typing import Any
+
+    from notion_client.client import AsyncClient
+
+
+class AsyncEndpoint:
+    def __init__(self, parent: AsyncClient) -> None:
+        self.parent = parent
+
+
+class BlocksChildrenEndpoint(AsyncEndpoint):
+    async def append(self, block_id: str, **kwargs: Any) -> Any:
+        """Create and append new children blocks to the block using the ID specified.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/patch-block-children)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path=f"blocks/{block_id}/children",
+            method="PATCH",
+            body=pick(kwargs, "children"),
+            auth=kwargs.get("auth"),
+        )
+
+    async def list(self, block_id: str, **kwargs: Any) -> Any:
+        """Return a paginated array of child [block objects](https://developers.notion.com/reference/block) contained in the block.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/get-block-children)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path=f"blocks/{block_id}/children",
+            method="GET",
+            query=pick(kwargs, "start_cursor", "page_size"),
+            auth=kwargs.get("auth"),
+        )
+
+
+class BlocksEndpoint(AsyncEndpoint):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.children = BlocksChildrenEndpoint(*args, **kwargs)
+
+    async def retrieve(self, block_id: str, **kwargs: Any) -> Any:
+        """Retrieve a [Block object](https://developers.notion.com/reference/block) using the ID specified.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-block)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path=f"blocks/{block_id}", method="GET", auth=kwargs.get("auth")
+        )
+
+    async def update(self, block_id: str, **kwargs: Any) -> Any:
+        """Update the content for the specified `block_id` based on the block type.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/update-a-block)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path=f"blocks/{block_id}",
+            method="PATCH",
+            body=pick(
+                kwargs,
+                "embed",
+                "type",
+                "archived",
+                "bookmark",
+                "image",
+                "video",
+                "pdf",
+                "file",
+                "audio",
+                "code",
+                "equation",
+                "divider",
+                "breadcrumb",
+                "table_of_contents",
+                "link_to_page",
+                "table_row",
+                "heading_1",
+                "heading_2",
+                "heading_3",
+                "paragraph",
+                "bulleted_list_item",
+                "numbered_list_item",
+                "quote",
+                "to_do",
+                "toggle",
+                "template",
+                "callout",
+                "synced_block",
+                "table",
+            ),
+            auth=kwargs.get("auth"),
+        )
+
+    def delete(self, block_id: str, **kwargs: Any) -> Any:
+        """Set a [Block object](https://developers.notion.com/reference/block), including page blocks, to `archived: true`.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/delete-a-block)*
+        """  # noqa: E501
+        return self.parent.request(
+            path=f"blocks/{block_id}",
+            method="DELETE",
+            auth=kwargs.get("auth"),
+        )
+
+
+class DatabasesEndpoint(AsyncEndpoint):
+    async def list(self, **kwargs: Any) -> Any:  # pragma: no cover
+        """List all [Databases](https://developers.notion.com/reference/database) shared with the authenticated integration.
+
+        > ⚠️  **Deprecated endpoint**
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/get-databases)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path="databases",
+            method="GET",
+            query=pick(kwargs, "start_cursor", "page_size"),
+            auth=kwargs.get("auth"),
+        )
+
+    async def query(self, database_id: str, **kwargs: Any) -> Any:
+        """Get a list of [Pages](https://developers.notion.com/reference/page) contained in the database.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/post-database-query)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path=f"databases/{database_id}/query",
+            method="POST",
+            query=pick(kwargs, "filter_properties"),
+            body=pick(kwargs, "filter", "sorts", "start_cursor", "page_size"),
+            auth=kwargs.get("auth"),
+        )
+
+    async def retrieve(self, database_id: str, **kwargs: Any) -> Any:
+        """Retrieve a [Database object](https://developers.notion.com/reference/database) using the ID specified.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-database)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path=f"databases/{database_id}", method="GET", auth=kwargs.get("auth")
+        )
+
+    async def create(self, **kwargs: Any) -> Any:
+        """Create a database as a subpage in the specified parent page.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/create-a-database)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path="databases",
+            method="POST",
+            body=pick(
+                kwargs, "parent", "title", "properties", "icon", "cover", "is_inline"
+            ),
+            auth=kwargs.get("auth"),
+        )
+
+    async def update(self, database_id: str, **kwargs: Any) -> Any:
+        """Update an existing database as specified by the parameters.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/update-a-database)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path=f"databases/{database_id}",
+            method="PATCH",
+            body=pick(kwargs, "properties", "title", "icon", "cover", "is_inline"),
+            auth=kwargs.get("auth"),
+        )
+
+
+class PagesPropertiesEndpoint(AsyncEndpoint):
+    async def retrieve(self, page_id: str, property_id: str, **kwargs: Any) -> Any:
+        """Retrieve a `property_item` object for a given `page_id` and `property_id`.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-page-property)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path=f"pages/{page_id}/properties/{property_id}",
+            method="GET",
+            auth=kwargs.get("auth"),
+            query=pick(kwargs, "start_cursor", "page_size"),
+        )
+
+
+class PagesEndpoint(AsyncEndpoint):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.properties = PagesPropertiesEndpoint(*args, **kwargs)
+
+    async def create(self, **kwargs: Any) -> Any:
+        """Create a new page in the specified database or as a child of an existing page.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/post-page)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path="pages",
+            method="POST",
+            body=pick(kwargs, "parent", "properties", "children", "icon", "cover"),
+            auth=kwargs.get("auth"),
+        )
+
+    async def retrieve(self, page_id: str, **kwargs: Any) -> Any:
+        """Retrieve a [Page object](https://developers.notion.com/reference/page) using the ID specified.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-page)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path=f"pages/{page_id}", method="GET", auth=kwargs.get("auth")
+        )
+
+    async def update(self, page_id: str, **kwargs: Any) -> Any:
+        """Update [page property values](https://developers.notion.com/reference/page#property-value-object) for the specified page.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/patch-page)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path=f"pages/{page_id}",
+            method="PATCH",
+            body=pick(kwargs, "archived", "properties", "icon", "cover"),
+            auth=kwargs.get("auth"),
+        )
+
+
+class SearchEndpoint(AsyncEndpoint):
+    async def __call__(self, **kwargs: Any) -> Any:
+        """Search all pages and child pages that are shared with the integration.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/post-search)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path="search",
+            method="POST",
+            body=pick(kwargs, "query", "sort", "filter", "start_cursor", "page_size"),
+            auth=kwargs.get("auth"),
+        )
+
+
+class CommentsEndpoint(AsyncEndpoint):
+    async def create(self, **kwargs: Any) -> Any:
+        """Create a new comment in the specified page or existing discussion thread.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/create-a-comment)*
+        """  # noqa: E501
+        return self.parent.request(
+            path="comments",
+            method="POST",
+            body=pick(kwargs, "parent", "discussion_id", "rich_text"),
+            auth=kwargs.get("auth"),
+        )
+
+    async def list(self, **kwargs: Any) -> Any:
+        """Retrieve a list of un-resolved [Comment objects](https://developers.notion.com/reference/comment-object) from the specified block.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-comment)*
+        """  # noqa: E501
+        return await self.parent.request(
+            path="comments",
+            method="GET",
+            query=pick(kwargs, "block_id", "start_cursor", "page_size"),
+            auth=kwargs.get("auth"),
+        )
+
+
+class UsersEndpoint(AsyncEndpoint):
+    async def list(self, **kwargs: Any) -> list[UserObjectResponse]:
+        """Return a paginated list of [Users](https://developers.notion.com/reference/user) for the workspace.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/get-users)*
+        """  # noqa: E501
+        users = await self.parent.request(
+            path="users",
+            method="GET",
+            query=pick(kwargs, "start_cursor", "page_size"),
+            auth=kwargs.get("auth"),
+        )
+
+        return cast(list[UserObjectResponse], users)
+
+    async def retrieve(self, user_id: str, **kwargs: Any) -> UserObjectResponse:
+        """Retrieve a [User](https://developers.notion.com/reference/user) using the ID specified.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/get-user)*
+        """  # noqa: E501
+        user = await self.parent.request(
+            path=f"users/{user_id}", method="GET", auth=kwargs.get("auth")
+        )
+
+        return cast(UserObjectResponse, user)
+
+    async def me(self, **kwargs: Any) -> BotUserObjectResponse:
+        """Retrieve the bot [User](https://developers.notion.com/reference/user) associated with the API token.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/get-self)*
+        """  # noqa: E501
+        bot = await self.parent.request(
+            path="users/me", method="GET", auth=kwargs.get("auth")
+        )
+
+        return cast(BotUserObjectResponse, bot)