Feat threads

docs/docs/usage/threads-graph-api.md

@@ -0,0 +1,43 @@
+## Introduction
+
+You may use the Threads API to enable people to create and publish content on a person’s behalf on Threads, and to
+display those posts within your app solely to the person who created it.
+
+## How to use
+
+Just like the base `Graph API`.
+
+The following code snippet shows how to perform an OAuth flow with the Threads API:
+
+```python
+from pyfacebook import ThreadsGraphAPI
+
+api = ThreadsGraphAPI(
+    app_id="Your app id",
+    app_secret="Your app secret",
+    oauth_flow=True,
+    redirect_uri="Your callback domain",
+    scope=["threads_basic", "threads_content_publish", "threads_read_replies", "threads_manage_replies",
+           "threads_manage_insights"]
+)
+
+# Got authorization url
+api.get_authorization_url()
+# https://threads.net/oauth/authorize?response_type=code&client_id=app_id&redirect_uri=https%3A%2F%2Fexample.com%2Fcallback&scope=threads_basic%2Cthreads_content_publish%2Cthreads_read_replies%2Cthreads_manage_replies%2Cthreads_manage_insights&state=PyFacebook
+
+# Once the user has authorized your app, you will get the redirected URL.
+# like `https://example.com/callback?code=AQBZzYhLZB&state=PyFacebook#_`
+token = api.exchange_user_access_token(response="Your response url")
+print(token)
+# {'access_token': 'access_token', 'user_id': 12342412}
+```
+
+After those steps, you can use the `api` object to call the Threads API.
+
+For example:
+
+```python
+api.get_object(object_id="me", fields=["id"])
+
+# {'id': '12342412'}
+```

docs/mkdocs.yml

@@ -33,6 +33,7 @@ nav:
   - Usage:
       - Graph API: usage/graph-api.md
       - Server-Sent Events: usage/server-send-events.md
+      - Threads Graph API: usage/threads-graph-api.md
   - Changelog: CHANGELOG.md
   - Previous Version Docs:
       - README: previous-version/readme.md

pyfacebook/__init__.py

@@ -1,7 +1,12 @@
 from pyfacebook.models import *
 from pyfacebook.ratelimit import RateLimitHeader, RateLimit, PercentSecond
 from pyfacebook.exceptions import PyFacebookException, FacebookError, LibraryError
-from pyfacebook.api import GraphAPI, BasicDisplayAPI, ServerSentEventAPI
+from pyfacebook.api import (
+    GraphAPI,
+    BasicDisplayAPI,
+    ThreadsGraphAPI,
+    ServerSentEventAPI,
+)
 from pyfacebook.api.facebook.client import FacebookApi
 from pyfacebook.api.instagram_business.client import IGBusinessApi
 from pyfacebook.api.instagram_basic.client import IGBasicDisplayApi

pyfacebook/api/__init__.py

@@ -1 +1 @@
-from .graph import GraphAPI, BasicDisplayAPI, ServerSentEventAPI
+from .graph import GraphAPI, BasicDisplayAPI, ThreadsGraphAPI, ServerSentEventAPI

pyfacebook/api/graph.py

@@ -8,7 +8,7 @@
 import re
 import time
 from urllib.parse import parse_qsl, urlparse
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Union
 from warnings import warn
 
 import requests
@@ -505,7 +505,7 @@ def delete_object(
     def _get_oauth_session(
         self,
         redirect_uri: Optional[str] = None,
-        scope: Optional[List[str]] = None,
+        scope: Optional[Union[List[str], str]] = None,
         state: Optional[str] = None,
         **kwargs,
     ) -> OAuth2Session:
@@ -869,6 +869,129 @@ def debug_token(self, input_token: str, access_token: Optional[str] = None) -> d
         raise LibraryError({"message": "Method not support"})
 
 
+class ThreadsGraphAPI(GraphAPI):
+    GRAPH_URL = "https://graph.threads.net/"
+    DEFAULT_SCOPE = ["threads_basic"]
+    AUTHORIZATION_URL = "https://threads.net/oauth/authorize"
+    EXCHANGE_ACCESS_TOKEN_URL = "https://graph.threads.net/oauth/access_token"
+
+    VALID_API_VERSIONS = ["v1.0"]
+
+    @staticmethod
+    def fix_scope(scope: Optional[List[str]] = None):
+        """
+        Note: After tests, the api for threads only support for comma-separated list.
+
+        :param scope: A list of permission string to request from the person using your app.
+        :return: comma-separated scope string
+        """
+        return ",".join(scope) if scope else scope
+
+    def get_authorization_url(
+        self,
+        redirect_uri: Optional[str] = None,
+        scope: Optional[List[str]] = None,
+        state: Optional[str] = None,
+        url_kwargs: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> Tuple[str, str]:
+        """
+        Build authorization url to do oauth.
+        Refer: https://developers.facebook.com/docs/facebook-login/manually-build-a-login-flow
+
+        :param redirect_uri: The URL that you want to redirect the person logging in back to.
+            Note: Your redirect uri need be set to `Valid OAuth redirect URIs` items in App Dashboard.
+        :param scope: A list of permission string to request from the person using your app.
+        :param state: A CSRF token that will be passed to the redirect URL.
+        :param url_kwargs: Additional parameters for generate authorization url. like config_id.
+        :param kwargs: Additional parameters for oauth.
+        :return: URL to do oauth and state
+        """
+        if scope:
+            self.scope = scope
+        scope = self.fix_scope(self.scope)
+
+        session = self._get_oauth_session(
+            redirect_uri=redirect_uri, scope=scope, state=state, **kwargs
+        )
+        url_kwargs = {} if url_kwargs is None else url_kwargs
+        authorization_url, state = session.authorization_url(
+            url=self.authorization_url, **url_kwargs
+        )
+        return authorization_url, state
+
+    def exchange_user_access_token(
+        self,
+        response: str,
+        redirect_uri: Optional[str] = None,
+        scope: Optional[List[str]] = None,
+        state: Optional[str] = None,
+        **kwargs,
+    ) -> dict:
+        """
+        :param response: The redirect response url for authorize redirect
+        :param redirect_uri: Url for your redirect.
+        :param scope: A list of permission string to request from the person using your app.
+        :param state: A CSRF token that will be passed to the redirect URL.
+        :param kwargs: Additional parameters for oauth.
+        :return:
+        """
+        if scope:
+            self.scope = scope
+        scope = self.fix_scope(self.scope)
+
+        session = self._get_oauth_session(
+            redirect_uri=redirect_uri, scope=scope, state=state, **kwargs
+        )
+
+        session.fetch_token(
+            self.access_token_url,
+            client_secret=self.app_secret,
+            authorization_response=response,
+            include_client_id=True,
+        )
+        self.access_token = session.access_token
+
+        return session.token
+
+    def exchange_long_lived_user_access_token(self, access_token=None) -> dict:
+        """
+        Generate long-lived token by short-lived token, Long-lived token generally lasts about 60 days.
+
+        :param access_token: Short-lived user access token
+        :return: Long-lived user access token info.
+        """
+        if access_token is None:
+            access_token = self.access_token
+        args = {
+            "grant_type": "th_exchange_token",
+            "client_id": self.app_id,
+            "client_secret": self.app_secret,
+            "access_token": access_token,
+        }
+
+        resp = self._request(
+            url=self.access_token_url,
+            args=args,
+            auth_need=False,
+        )
+        data = self._parse_response(resp)
+        return data
+
+    def refresh_access_token(self, access_token: str):
+        """
+        :param access_token: The valid (unexpired) long-lived Instagram User Access Token that you want to refresh.
+        :return: New access token.
+        """
+        args = {"grant_type": "th_refresh_token", "access_token": access_token}
+        resp = self._request(
+            url="refresh_access_token",
+            args=args,
+        )
+        data = self._parse_response(resp)
+        return data
+
+
 class ServerSentEventAPI:
     """
     Notice: Server-Sent Events are deprecated and will be removed December 31, 2023.

testdata/base/threads_user_long_lived_token.json

@@ -0,0 +1 @@
+{"access_token": "THQVJ...","token_type": "bearer", "expires_in": 5183944}

testdata/base/threads_user_token.json

@@ -0,0 +1 @@
+{"access_token": "THQVJ...","user_id": 12345678}

tests/test_threads_graph_api.py

@@ -0,0 +1,59 @@
+"""
+    tests for threads graph api
+"""
+
+import responses
+
+from pyfacebook import ThreadsGraphAPI
+
+
+def test_threads_get_authorization_url():
+    api = ThreadsGraphAPI(app_id="id", app_secret="secret", oauth_flow=True)
+
+    url, state = api.get_authorization_url(scope=["threads_basic"])
+    assert (
+        url
+        == "https://threads.net/oauth/authorize?response_type=code&client_id=id&redirect_uri=https%3A%2F%2Flocalhost%2F&scope=threads_basic&state=PyFacebook"
+    )
+
+
+def test_threads_exchange_user_access_token(helpers):
+    api = ThreadsGraphAPI(app_id="id", app_secret="secret", oauth_flow=True)
+
+    resp = "https://localhost/?code=code&state=PyFacebook#_"
+
+    with responses.RequestsMock() as m:
+        m.add(
+            method=responses.POST,
+            url=api.EXCHANGE_ACCESS_TOKEN_URL,
+            json=helpers.load_json("testdata/base/threads_user_token.json"),
+        )
+
+        r = api.exchange_user_access_token(response=resp, scope=["threads_basic"])
+        assert r["access_token"] == "THQVJ..."
+
+
+def test_threads_exchange_long_lived_user_access_token(helpers):
+    api = ThreadsGraphAPI(app_id="id", app_secret="secret", access_token="token")
+    with responses.RequestsMock() as m:
+        m.add(
+            method=responses.GET,
+            url=f"https://graph.threads.net/oauth/access_token",
+            json=helpers.load_json("testdata/base/threads_user_long_lived_token.json"),
+        )
+
+        r = api.exchange_long_lived_user_access_token()
+        assert r["access_token"] == "THQVJ..."
+
+
+def test_threads_refresh_access_token(helpers):
+    api = ThreadsGraphAPI(app_id="id", app_secret="secret", access_token="token")
+    with responses.RequestsMock() as m:
+        m.add(
+            method=responses.GET,
+            url=f"https://graph.threads.net/refresh_access_token",
+            json=helpers.load_json("testdata/base/threads_user_long_lived_token.json"),
+        )
+
+        r = api.refresh_access_token(access_token=api.access_token)
+        assert r["access_token"] == "THQVJ..."