Adding OAuth revoke and introspect endpoints

.github/workflows/test.yml

@@ -13,7 +13,7 @@ jobs:
 
     strategy:
       matrix:
-        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13"]
 
     steps:
 

README.md

@@ -250,7 +250,7 @@ at the end of the session.
 
 This package supports the following minimum versions:
 
-* Python >= 3.7
+* Python >= 3.8
 * httpx >= 0.23.0
 
 Earlier versions may still work, but we encourage people building new applications

notion_client/api_endpoints.py

@@ -325,3 +325,41 @@ def list(self, **kwargs: Any) -> SyncAsync[Any]:
             query=pick(kwargs, "block_id", "start_cursor", "page_size"),
             auth=kwargs.get("auth"),
         )
+
+
+class OAuthEndpoint(Endpoint):
+    def token(self, **kwargs: Any) -> SyncAsync[Any]:
+        """Creates an access token that a third-party service can use to authenticate with Notion.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/create-a-token)*
+        """  # noqa: E501
+        return self.parent.request(
+            path="oauth/token",
+            method="POST",
+            body=pick(kwargs, "grant_type", "code", "redirect_uri"),
+            auth=kwargs.get("auth"),
+        )
+
+    def introspect(self, **kwargs: Any) -> SyncAsync[Any]:
+        """Get a token's active status, scope, and issued time.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/introspect-token)*
+        """  # noqa: E501
+        return self.parent.request(
+            path="oauth/introspect",
+            method="POST",
+            body=pick(kwargs, "token"),
+            auth=kwargs.get("auth"),
+        )
+
+    def revoke(self, **kwargs: Any) -> SyncAsync[Any]:
+        """Revoke an access token.
+
+        *[🔗 Endpoint documentation](https://developers.notion.com/reference/revoke-token)*
+        """  # noqa: E501
+        return self.parent.request(
+            path="oauth/revoke",
+            method="POST",
+            body=pick(kwargs, "token"),
+            auth=kwargs.get("auth"),
+        )

notion_client/client.py

@@ -9,13 +9,16 @@
 import httpx
 from httpx import Request, Response
 
+import base64
+
 from notion_client.api_endpoints import (
     BlocksEndpoint,
     CommentsEndpoint,
     DatabasesEndpoint,
     PagesEndpoint,
     SearchEndpoint,
     UsersEndpoint,
+    OAuthEndpoint,
 )
 from notion_client.errors import (
     APIResponseError,
@@ -24,7 +27,7 @@
     is_api_error_code,
 )
 from notion_client.logging import make_console_logger
-from notion_client.typing import SyncAsync
+from notion_client.typing import SyncAsync, OAuthHeader
 
 
 @dataclass
@@ -77,6 +80,7 @@ def __init__(
         self.pages = PagesEndpoint(self)
         self.search = SearchEndpoint(self)
         self.comments = CommentsEndpoint(self)
+        self.oauth = OAuthEndpoint(self)
 
     @property
     def client(self) -> Union[httpx.Client, httpx.AsyncClient]:
@@ -102,11 +106,21 @@ def _build_request(
         path: str,
         query: Optional[Dict[Any, Any]] = None,
         body: Optional[Dict[Any, Any]] = None,
-        auth: Optional[str] = None,
+        auth: Optional[Union[str, OAuthHeader]] = None,
     ) -> Request:
         headers = httpx.Headers()
         if auth:
-            headers["Authorization"] = f"Bearer {auth}"
+            # At runtime the TypedDict is the same type as a regular Dict
+            if isinstance(auth, Dict):
+                client_id = auth["client_id"]
+                client_secret = auth["client_secret"]
+                unencoded_credential = f"{client_id}:{client_secret}"
+                encoded_credential = base64.b64encode(
+                    unencoded_credential.encode()
+                ).decode("utf-8")
+                headers["Authorization"] = f'Basic "{encoded_credential}"'
+            else:
+                headers["Authorization"] = f"Bearer {auth}"
         self.logger.info(f"{method} {self.client.base_url}{path}")
         self.logger.debug(f"=> {query} -- {body}")
         return self.client.build_request(

notion_client/typing.py

@@ -1,5 +1,10 @@
 """Custom type definitions for notion-sdk-py."""
-from typing import Awaitable, TypeVar, Union
+from typing import Awaitable, TypeVar, Union, TypedDict
 
 T = TypeVar("T")
 SyncAsync = Union[T, Awaitable[T]]
+
+
+class OAuthHeader(TypedDict):
+    client_id: str
+    client_secret: str

setup.py

@@ -21,7 +21,6 @@ def get_description():
         "httpx >= 0.23.0",
     ],
     classifiers=[
-        "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
         "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",

tests/cassettes/test_client_request_oauth.yaml

@@ -0,0 +1,75 @@
+interactions:
+- request:
+    body: ''
+    headers:
+      accept:
+      - '*/*'
+      accept-encoding:
+      - gzip, deflate
+      connection:
+      - keep-alive
+      content-length:
+      - '0'
+      host:
+      - api.notion.com
+      notion-version:
+      - '2022-06-28'
+    method: POST
+    uri: https://api.notion.com/v1/oauth/introspect
+  response:
+    content: '{"error":"invalid_client","request_id":"141bd2e5-09c1-4697-b80b-5fd2fd4dc45b"}'
+    headers: {}
+    http_version: HTTP/1.1
+    status_code: 401
+- request:
+    body: ''
+    headers:
+      accept:
+      - '*/*'
+      accept-encoding:
+      - gzip, deflate
+      authorization:
+      - ntn_...
+      connection:
+      - keep-alive
+      content-length:
+      - '0'
+      host:
+      - api.notion.com
+      notion-version:
+      - '2022-06-28'
+    method: POST
+    uri: https://api.notion.com/v1/oauth/introspect
+  response:
+    content: '{"error":"invalid_client","request_id":"f678e9ff-7a4e-4ba7-a74e-d9edb7b81047"}'
+    headers: {}
+    http_version: HTTP/1.1
+    status_code: 401
+- request:
+    body: '{"token": "ntn_..."}'
+    headers:
+      accept:
+      - '*/*'
+      accept-encoding:
+      - gzip, deflate
+      authorization:
+      - Basic "Base64Encoded($client_id:$client_secret)"
+      connection:
+      - keep-alive
+      content-length:
+      - '63'
+      content-type:
+      - application/json
+      host:
+      - api.notion.com
+      notion-version:
+      - '2022-06-28'
+    method: POST
+    uri: https://api.notion.com/v1/oauth/introspect
+  response:
+    content: '{"active":true,"scope":"read_content insert_content update_content read_user_with_email
+      read_user_without_email","iat":1742416470043,"request_id":"498e8bad-8927-4dd9-9b82-bf7e051f86d4"}'
+    headers: {}
+    http_version: HTTP/1.1
+    status_code: 200
+version: 1

tests/cassettes/test_introspect_token.yaml

@@ -0,0 +1,29 @@
+interactions:
+- request:
+    body: '{"token": "ntn_..."}'
+    headers:
+      accept:
+      - '*/*'
+      accept-encoding:
+      - gzip, deflate
+      authorization:
+      - Basic "Base64Encoded($client_id:$client_secret)"
+      connection:
+      - keep-alive
+      content-length:
+      - '63'
+      content-type:
+      - application/json
+      host:
+      - api.notion.com
+      notion-version:
+      - '2022-06-28'
+    method: POST
+    uri: https://api.notion.com/v1/oauth/introspect
+  response:
+    content: '{"active":true,"scope":"read_content insert_content update_content read_user_with_email
+      read_user_without_email","iat":1742416470043,"request_id":"57f9e373-a28b-4b8f-b2dc-c0d687f6f743"}'
+    headers: {}
+    http_version: HTTP/1.1
+    status_code: 200
+version: 1

tests/cassettes/test_revoke_token.yaml

@@ -0,0 +1,28 @@
+interactions:
+- request:
+    body: '{"token": "ntn_..."}'
+    headers:
+      accept:
+      - '*/*'
+      accept-encoding:
+      - gzip, deflate
+      authorization:
+      - Basic "Base64Encoded($client_id:$client_secret)"
+      connection:
+      - keep-alive
+      content-length:
+      - '63'
+      content-type:
+      - application/json
+      host:
+      - api.notion.com
+      notion-version:
+      - '2022-06-28'
+    method: POST
+    uri: https://api.notion.com/v1/oauth/revoke
+  response:
+    content: '{"request_id":"a6dcf82b-8a97-48af-b851-8b9b3559ed69"}'
+    headers: {}
+    http_version: HTTP/1.1
+    status_code: 200
+version: 1

tests/cassettes/test_token.yaml

@@ -0,0 +1,28 @@
+interactions:
+- request:
+    body: '{"grant_type": "authorization_code", "code": "...", "redirect_uri": "http://..."}'
+    headers:
+      accept:
+      - '*/*'
+      accept-encoding:
+      - gzip, deflate
+      authorization:
+      - Basic "Base64Encoded($client_id:$client_secret)"
+      connection:
+      - keep-alive
+      content-length:
+      - '134'
+      content-type:
+      - application/json
+      host:
+      - api.notion.com
+      notion-version:
+      - '2022-06-28'
+    method: POST
+    uri: https://api.notion.com/v1/oauth/token
+  response:
+    content: '{"access_token":"...","token_type":"...","bot_id":"...","workspace_name":"...","workspace_icon":"...","workspace_id":"...","owner":"...","duplicated_template_id":"...","request_id":"..."}'
+    headers: {}
+    http_version: HTTP/1.1
+    status_code: 200
+version: 1

tests/conftest.py

@@ -2,8 +2,10 @@
 import re
 from datetime import datetime
 from typing import Optional
+import json
 
 import pytest
+from vcr.request import Request
 
 from notion_client import AsyncClient, Client
 
@@ -14,13 +16,50 @@ def remove_headers(response: dict):
         response["headers"] = {}
         return response
 
+    def scrub_requests(request: Request):
+        if request.body:
+            body_str = request.body.decode("utf-8")
+            body_json = json.loads(body_str)
+            if "token" in body_json:
+                body_json["token"] = "ntn_..."
+            if "code" in body_json:
+                body_json["code"] = "..."
+            if "redirect_uri" in body_json:
+                body_json["redirect_uri"] = "http://..."
+            request.body = json.dumps(body_json).encode("utf-8")
+        return request
+
+    def scrub_response(response: dict):
+        if "content" in response:
+            content = response["content"]
+            # Like the case tests/cassettes/test_api_async_request_bad_request_error.yaml, where the response is just a string, not JSON
+            # We don't want to raise an error here because the response is not JSON and that is ok
+            if "{" not in content:
+                return response
+            content_json = json.loads(content)
+            if "access_token" in content_json:
+                response["content"] = json.dumps(
+                    {key: "..." for key in content_json}, separators=(",", ":")
+                )
+        return response
+
+    # The VCR config requires the passing of the request parameter, despite the face that it is not used
+    # (https://vcrpy.readthedocs.io/en/latest/advanced.html#advanced-use-of-filter-headers-filter-query-parameters-and-filter-post-data-parameters)
+    def scrub_auth_header(key: str, value: str, request: Optional[Request]):
+        if key == "authorization":
+            if value.startswith("Bearer "):
+                return "ntn_..."
+            elif value.startswith("Basic "):
+                return 'Basic "Base64Encoded($client_id:$client_secret)"'
+
     return {
         "filter_headers": [
-            ("authorization", "ntn_..."),
+            ("authorization", scrub_auth_header),
             ("user-agent", None),
             ("cookie", None),
         ],
-        "before_record_response": remove_headers,
+        "before_record_request": scrub_requests,
+        "before_record_response": (remove_headers, scrub_response),
         "match_on": ["method", "remove_page_id_for_matches"],
     }
 
@@ -40,6 +79,26 @@ def token() -> str:
     return os.environ.get("NOTION_TOKEN")
 
 
+@pytest.fixture(scope="session")
+def code() -> str:
+    return os.environ.get("NOTION_CODE")
+
+
+@pytest.fixture(scope="session")
+def redirect_uri() -> str:
+    return os.environ.get("NOTION_REDIRECT_URI")
+
+
+@pytest.fixture(scope="session")
+def client_id() -> str:
+    return os.environ.get("NOTION_CLIENT_ID")
+
+
+@pytest.fixture(scope="session")
+def client_secret() -> str:
+    return os.environ.get("NOTION_CLIENT_SECRET")
+
+
 @pytest.fixture(scope="module", autouse=True)
 def parent_page_id(vcr) -> str:
     """this is the ID of the Notion page where the tests will be executed