diff --git a/docs/docs/usage/threads-graph-api.md b/docs/docs/usage/threads-graph-api.md new file mode 100644 index 00000000..e487f27d --- /dev/null +++ b/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'} +``` diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index ab4c3151..ad79afa4 100644 --- a/docs/mkdocs.yml +++ b/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 diff --git a/pyfacebook/__init__.py b/pyfacebook/__init__.py index 5a0aefc4..4a1dc535 100644 --- a/pyfacebook/__init__.py +++ b/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 diff --git a/pyfacebook/api/__init__.py b/pyfacebook/api/__init__.py index cf55e760..e0f9d3ed 100644 --- a/pyfacebook/api/__init__.py +++ b/pyfacebook/api/__init__.py @@ -1 +1 @@ -from .graph import GraphAPI, BasicDisplayAPI, ServerSentEventAPI +from .graph import GraphAPI, BasicDisplayAPI, ThreadsGraphAPI, ServerSentEventAPI diff --git a/pyfacebook/api/graph.py b/pyfacebook/api/graph.py index ad79d28c..cfe76e83 100644 --- a/pyfacebook/api/graph.py +++ b/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. diff --git a/testdata/base/threads_user_long_lived_token.json b/testdata/base/threads_user_long_lived_token.json new file mode 100644 index 00000000..76305f32 --- /dev/null +++ b/testdata/base/threads_user_long_lived_token.json @@ -0,0 +1 @@ +{"access_token": "THQVJ...","token_type": "bearer", "expires_in": 5183944} \ No newline at end of file diff --git a/testdata/base/threads_user_token.json b/testdata/base/threads_user_token.json new file mode 100644 index 00000000..6ae1785e --- /dev/null +++ b/testdata/base/threads_user_token.json @@ -0,0 +1 @@ +{"access_token": "THQVJ...","user_id": 12345678} \ No newline at end of file diff --git a/tests/test_threads_graph_api.py b/tests/test_threads_graph_api.py new file mode 100644 index 00000000..eda23aac --- /dev/null +++ b/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..."