python: support client credentials

nodejs/scripts/get_access_token.js

@@ -30,13 +30,17 @@ import { Scope, print_scopes } from '../src/oauth_scope.js';
  *    OAuth scopes, allows experimentation with different sets of scopes. Specifying scopes prevents
  *    the access token from being read from the environment or file system, and forces the use of
  *    the browser-based OAuth process.
+ *  -c / --client-credentials:
+ *    This option is used to request an access token for the user account
+ *    associated with the PINTEREST_APP_ID. Use this option to avoid the need
+ *    to do the manual part of the OAuth process with at web browser.
  */
 async function main(argv) {
   const parser = new ArgumentParser({ description: 'Get Pinterest OAuth token' });
   parser.add_argument('-w', '--write', { action: 'store_true', help: 'write access token to file' });
   parser.add_argument('-ct', '--cleartext', { action: 'store_true', help: 'print the token in clear text' });
   parser.add_argument('-s', '--scopes', { help: 'comma separated list of scopes or "help"' });
-  parser.add_argument('-c', '--client_credentials', { action: 'store_true', help: 'use client credentials' });
+  parser.add_argument('-c', '--client_credentials', { action: 'store_true', help: 'access the application user account' });
   common_arguments(parser);
   const args = parser.parse_args(argv);
 

python/scripts/copy_board.py

@@ -158,8 +158,7 @@ def copy_pin(pin, pin_data, target_board_id, target_section_id=None):
 
     if args.all_boards:  # copy all boards for the source user
         user = User(api_config, source_token)
-        user_data = user.get()
-        boards = user.get_boards(user_data)
+        boards = user.get_boards()
         source_board = Board(
             None, api_config, source_token
         )  # board_id set in loop below

python/scripts/delete_board.py

@@ -46,7 +46,7 @@ def main(argv=[]):
     if args.all_boards:  # delete all boards for the user
         user = User(api_config, access_token)
         user_data = user.get()
-        boards = user.get_boards(user_data)
+        boards = user.get_boards()
         confirmation = f"Delete all boards for {user_data['username']}"
     else:  # copy just the board designated by board_id
         deletion_board = Board(args.board_id, api_config, access_token)

python/scripts/get_access_token.py

@@ -45,6 +45,10 @@ def main(argv=[]):
        allows experimentation with different sets of scopes. Specifying scopes prevents
        the access token from being read from the environment or file system, and forces
        the use of the browser-based OAuth process.
+     -c / --client-credentials:
+       This option is used to request an access token for the user account
+       associated with the PINTEREST_APP_ID. Use this option to avoid the need
+       to do the manual part of the OAuth process with at web browser.
     """
     parser = argparse.ArgumentParser(description="Get Pinterest OAuth token")
     parser.add_argument(
@@ -56,6 +60,12 @@ def main(argv=[]):
     parser.add_argument(
         "-s", "--scopes", help="comma separated list of scopes or 'help'"
     )
+    parser.add_argument(
+        "-c",
+        "--client-credentials",
+        action="store_true",
+        help="access the application user account",
+    )
     common_arguments(parser)
     args = parser.parse_args(argv)
 
@@ -70,10 +80,10 @@ def main(argv=[]):
         # use the comma-separated list of scopes passed as a command-line argument
         scope_list = args.scopes.split(",")
         scopes = list(map(lookup_scope, scope_list))
-        access_token.oauth(scopes=scopes)
+        access_token.oauth(scopes=scopes, client_credentials=args.client_credentials)
     else:
         try:
-            access_token.fetch()
+            access_token.fetch(client_credentials=args.client_credentials)
         except ValueError as err:
             # ValueError indicates that something was wrong with the arguments
             parser.error(err)
@@ -98,11 +108,13 @@ def main(argv=[]):
         print("writing access token")
         access_token.write()
 
-    # Use the access token to get information about the user. The purpose of this
-    # call is to verify that the access token is working.
-    user = User(api_config, access_token)
-    user_data = user.get()
-    user.print_summary(user_data)
+    # Remove this conditional when GET /v5/user_account works with client credentials
+    if not args.client_credentials:
+        # Use the access token to get information about the user. The purpose of this
+        # call is to verify that the access token is working.
+        user = User(api_config, access_token)
+        user_data = user.get()
+        user.print_summary(user_data)
 
 
 # If this script is being called from the command line, call the main function

python/scripts/get_user_boards.py

@@ -55,17 +55,14 @@ def main(argv=[]):
     access_token = AccessToken(api_config, name=args.access_token)
     access_token.fetch(scopes=[Scope.READ_USERS, Scope.READ_BOARDS])
 
-    # use the access token to get information about the user
-    user = User(api_config, access_token)
-    user_data = user.get()
-
     # get information about all of the boards in the user's profile
+    user = User(api_config, access_token)
     query_parameters = {"page_size": args.page_size}
     if args.include_empty:
         query_parameters["include_empty"] = args.include_empty
     if args.include_archived:
         query_parameters["include_archived"] = args.include_archived
-    board_iterator = user.get_boards(user_data, query_parameters)
+    board_iterator = user.get_boards(query_parameters)
     user.print_multiple(args.page_size, "board", Board, board_iterator)
 
 

python/scripts/get_user_pins.py

@@ -34,14 +34,9 @@ def main(argv=[]):
     access_token = AccessToken(api_config, name=args.access_token)
     access_token.fetch(scopes=[Scope.READ_USERS, Scope.READ_PINS, Scope.READ_BOARDS])
 
-    # use the access token to get information about the user
-    user = User(api_config, access_token)
-    user_data = user.get()
-
     # get information about all of the pins in the user's profile
-    pin_iterator = user.get_pins(
-        user_data, query_parameters={"page_size": args.page_size}
-    )
+    user = User(api_config, access_token)
+    pin_iterator = user.get_pins(query_parameters={"page_size": args.page_size})
     user.print_multiple(args.page_size, "pin", Pin, pin_iterator)
 
 

python/src/access_token.py

@@ -26,7 +26,7 @@ def __init__(self, api_config, name=None):
         b64auth = base64.b64encode(auth.encode("ascii")).decode("ascii")
         self.auth_headers = {"Authorization": "Basic " + b64auth}
 
-    def fetch(self, scopes=None, refreshable=True):
+    def fetch(self, scopes=None, client_credentials=False):
         """
         This method tries to make it as easy as possible for a developer
         to start using an OAuth access token. It fetches the access token
@@ -53,7 +53,7 @@ def fetch(self, scopes=None, refreshable=True):
         except Exception:
             print(f"reading {self.name} failed, trying oauth")
 
-        self.oauth(scopes=scopes, refreshable=refreshable)
+        self.oauth(scopes=scopes, client_credentials=client_credentials)
 
     def from_environment(self):
         """
@@ -118,44 +118,55 @@ def hashed_refresh_token(self):
             raise RuntimeError("AccessToken does not have a refresh token")
         return hashlib.sha256(self.refresh_token.encode()).hexdigest()
 
-    def oauth(self, scopes=None, refreshable=True):
+    def _get_user_post_data(self, scopes):
+        """
+        When requesting an OAuth token for a user, the protocol requires going
+        through the process of getting an auth_code. This process allows the
+        user to approve the scopes requested by the application.
+        """
+        print("getting auth_code...")
+        auth_code = get_auth_code(self.api_config, scopes=scopes)
+        print(f"exchanging auth_code for {self.name}...")
+
+        # Generate POST data to exchange the auth_code (obtained by
+        # a redirect from the browser) for the access_token and a
+        # refresh_token.
+        return {
+            "code": auth_code,
+            "redirect_uri": self.api_config.redirect_uri,
+            "grant_type": "authorization_code",
+        }
+
+    def _get_client_post_data(self, scopes):
+        """
+        When requesting an OAuth token for the client, no auth_code is required
+        because the user is the same as the owner of the client.
+        """
+        print("getting access token using client credentials...")
+        return {
+            "grant_type": "client_credentials",
+            "scope": ",".join(list(map(lambda scope: scope.value, scopes))),
+        }
+
+    def oauth(self, scopes=None, client_credentials=False):
         """
         Execute the OAuth 2.0 process for obtaining an access token.
         For more information, see IETF RFC 6749: https://tools.ietf.org/html/rfc6749
         and https://developers.pinterest.com/docs/getting-started/authentication/
-
-        For v5, scopes are required and tokens must be refreshable.
         """
         if not scopes:
             scopes = [Scope.READ_USERS, Scope.READ_PINS, Scope.READ_BOARDS]
             print(
-                "v5 requires scopes for OAuth. setting to default: "
+                "OAuth scopes required. Setting to default: "
                 f"{','.join(list(map(lambda scope: scope.value, scopes)))}"
             )
 
-        if not refreshable:
-            raise ValueError(
-                "Pinterest API v5 only provides refreshable OAuth access tokens"
-            )
-
-        print("getting auth_code...")
-        auth_code = get_auth_code(
-            self.api_config, scopes=scopes, refreshable=refreshable
+        post_data = (
+            client_credentials
+            and self._get_client_post_data(scopes)
+            or self._get_user_post_data(scopes)
         )
-        print(f"exchanging auth_code for {self.name}...")
-        self.exchange_auth_code(auth_code)
 
-    def exchange_auth_code(self, auth_code):
-        """
-        Call the Pinterest API to exchange the auth_code (obtained by
-        a redirect from the browser) for the access_token and (if requested)
-        refresh_token.
-        """
-        post_data = {
-            "code": auth_code,
-            "redirect_uri": self.api_config.redirect_uri,
-            "grant_type": "authorization_code",
-        }
         if self.api_config.verbosity >= 2:
             print("POST", self.api_config.api_uri + "/v5/oauth/token")
             if self.api_config.verbosity >= 3:
@@ -170,7 +181,7 @@ def exchange_auth_code(self, auth_code):
 
         print("scope: " + unpacked["scope"])
         self.access_token = unpacked["access_token"]
-        self.refresh_token = unpacked["refresh_token"]
+        self.refresh_token = unpacked.get("refresh_token")
         self.scopes = unpacked["scope"]
 
     def refresh(self, continuous=False):

python/src/user.py

@@ -18,10 +18,10 @@ def print_summary(self, user_data):
         print("--------------------")
 
     # https://developers.pinterest.com/docs/api/v5/#operation/boards/list
-    def get_boards(self, user_data, query_parameters=None):
+    def get_boards(self, query_parameters=None):
         # the returned iterator handles API paging
         return self.get_iterator("/v5/boards", query_parameters)
 
     # getting all of a user's pins is not supported, so iterate through boards
-    def get_pins(self, user_data, query_parameters=None):
+    def get_pins(self, query_parameters=None):
         return self.get_iterator("/v5/pins", query_parameters)

python/tests/src/test_access_token.py

@@ -111,7 +111,7 @@ def test_access_token(self, rm, mock_get_auth_code):
 
         # mock does not figure out enum equality, so need to unpack arguments
         # of get_auth_code in order to check the default value for scopes.
-        self.assertTrue(mock_get_auth_code.mock_calls[0][2]["refreshable"])
+        self.assertNotIn("refreshable", mock_get_auth_code.mock_calls[0][2])
         scopes = mock_get_auth_code.mock_calls[0][2]["scopes"]
         # convert from Scope enum to values
         values = list(map(lambda scope: scope.value, scopes))
@@ -124,6 +124,14 @@ def test_access_token(self, rm, mock_get_auth_code):
             + "&grant_type=authorization_code",
         )
 
+        # verify OAuth with client credentials
+        access_token.oauth(client_credentials=True)
+        self.assertEqual(
+            rm.last_request.text,
+            "grant_type=client_credentials"
+            + "&scope=user_accounts%3Aread%2Cpins%3Aread%2Cboards%3Aread",
+        )
+
         rm.post(
             "https://test-api-uri/v5/oauth/token",
             request_headers={
@@ -203,12 +211,7 @@ def test_access_token(self, rm, mock_get_auth_code):
         )
         mock_get_auth_code.reset_mock()
         access_token = AccessToken(mock_api_config)
-        access_token.oauth(scopes=["test-scope-1", "test-scope-2"], refreshable=True)
+        access_token.oauth(scopes=["test-scope-1", "test-scope-2"])
         mock_get_auth_code.assert_called_once_with(
-            mock_api_config, scopes=["test-scope-1", "test-scope-2"], refreshable=True
+            mock_api_config, scopes=["test-scope-1", "test-scope-2"]
         )
-
-        with self.assertRaisesRegex(
-            ValueError, "Pinterest API v5 only provides refreshable OAuth access tokens"
-        ):
-            access_token.oauth(refreshable=False)

python/tests/src/test_user.py

@@ -25,18 +25,17 @@ def test_user_get_boards(self, mock_api_object_init, mock_api_object_get_iterato
 
         mock_api_object_get_iterator.return_value = "test_iterator"
         response = test_user.get_boards(
-            "test_user_data", query_parameters={"param1": "value1", "param2": "value2"}
+            query_parameters={"param1": "value1", "param2": "value2"}
         )
         mock_api_object_get_iterator.assert_called_once_with(
             "/v5/boards", {"param1": "value1", "param2": "value2"}
         )
         self.assertEqual(response, "test_iterator")
 
-        response = test_user.get_boards("test_user_data")
+        response = test_user.get_boards()
         mock_api_object_get_iterator.assert_called_with("/v5/boards", None)
 
         response = test_user.get_boards(
-            "test_user_data",
             query_parameters={
                 "param1": "value1",
             },
@@ -61,7 +60,7 @@ def test_user_get_pins(self, mock_api_object_get_iterator):
         expected_pins = ["board1_pin1", "board1_pin2", "board3_pin1"]
         test_user = User(mock_api_config, "test_access_token")
         for index, pin in enumerate(
-            test_user.get_pins("test_user_data", query_parameters={"param1": "value1"})
+            test_user.get_pins(query_parameters={"param1": "value1"})
         ):
             self.assertEqual(expected_pins[index], pin)