Tumblr method for fetching primary blog ID (used as user ID)

src/pardner/services/tumblr.py

@@ -1,3 +1,4 @@
+import json
 from typing import Any, Iterable, Optional, override
 
 from pardner.exceptions import UnsupportedRequestException
@@ -12,6 +13,7 @@ class TumblrTransferService(BaseTransferService):
     See API documentation: https://www.tumblr.com/docs/en/api/v2
     """
 
+    primary_blog_id: str | None = None
     _authorization_url = 'https://www.tumblr.com/oauth2/authorize'
     _base_url = 'https://api.tumblr.com/v2/'
     _token_url = 'https://api.tumblr.com/v2/oauth2/token'
@@ -23,7 +25,20 @@ def __init__(
         redirect_uri: str,
         state: Optional[str] = None,
         verticals: set[Vertical] = set(),
+        primary_blog_id: str | None = None,
     ) -> None:
+        """
+        Creates an instance of ``TumblrTransferService``.
+
+        :param client_id: Client identifier given by the OAuth provider upon registration.
+        :param client_secret: The ``client_secret`` paired to the ``client_id``.
+        :param redirect_uri: The registered callback URI.
+        :param state: State string used to prevent CSRF and identify flow.
+        :param verticals: The :class:`Vertical`s for which the transfer service has
+        appropriate scope to fetch.
+        :param primary_blog_id: Optionally, the primary blog ID of the data owner (the
+        user being authorized).
+        """
         super().__init__(
             service_name='Tumblr',
             client_id=client_id,
@@ -33,6 +48,7 @@ def __init__(
             supported_verticals={SocialPostingVertical},
             verticals=verticals,
         )
+        self.primary_blog_id = primary_blog_id
 
     @override
     def scope_for_verticals(self, verticals: Iterable[Vertical]) -> set[str]:
@@ -48,6 +64,40 @@ def fetch_token(
     ) -> dict[str, Any]:
         return super().fetch_token(code, authorization_response, include_client_id)
 
+    def fetch_primary_blog_id(self) -> str:
+        """
+        Fetches the primary blog ID from the data owner, which will be used as the
+        ``data_owner_id`` in the vertical model objects. If the ``primary_blog_id``
+        attribute on this class is already set, the method does not make a new request.
+
+        Note: "PrimaryBlogId" is not a vertical. This is used purely as a unique
+        identifier for the user, since Tumblr doesn't provide one by default.
+
+        :returns: the primary blog id.
+
+        :raises: :class:`ValueError`: if the primary blog ID could not be extracted from
+        the response.
+        """
+        if self.primary_blog_id:
+            return self.primary_blog_id
+        user_info = self._get_resource_from_path('user/info').json().get('response', {})
+        for blog_info in user_info.get('user', {}).get('blogs', []):
+            if (
+                isinstance(blog_info, dict)
+                and blog_info.get('primary')
+                and 'uuid' in blog_info
+                and isinstance(blog_info['uuid'], str)
+            ):
+                self.primary_blog_id = blog_info['uuid']
+                return blog_info['uuid']
+
+        raise ValueError(
+            'Failed to fetch primary blog id. Either manually set the _primary_blog_id '
+            'attribute or verify all the client credentials '
+            'and permissions are correct. Response from Tumblr: '
+            f'{json.dumps(user_info, indent=2)}'
+        )
+
     def fetch_social_posting_vertical(
         self,
         request_params: dict[str, Any] = {},

tests/test_transfer_services/test_tumblr.py

@@ -40,3 +40,35 @@ def test_fetch_social_posting_vertical(mocker, tumblr_transfer_service):
         oauth2_session_get.call_args.args[1]
         == 'https://api.tumblr.com/v2/user/dashboard'
     )
+
+
+def test_fetch_primary_blog_id_already_set(tumblr_transfer_service):
+    tumblr_transfer_service.primary_blog_id = 'existing-blog-id'
+    assert tumblr_transfer_service.fetch_primary_blog_id() == 'existing-blog-id'
+
+
+def test_fetch_primary_blog_id_success(mocker, tumblr_transfer_service):
+    response_object = mocker.MagicMock()
+    response_object.json.return_value = {
+        'response': {
+            'user': {
+                'blogs': [
+                    {'primary': False, 'uuid': 'secondary-blog-id'},
+                    {'primary': True, 'uuid': 'primary-blog-id', 'name': 'my-blog'},
+                    {'primary': False, 'uuid': 'another-secondary-id'},
+                ]
+            }
+        }
+    }
+    oauth2_session_get = mock_oauth2_session_get(mocker, response_object)
+
+    assert tumblr_transfer_service.fetch_primary_blog_id() == 'primary-blog-id'
+    assert tumblr_transfer_service.primary_blog_id == 'primary-blog-id'
+    assert oauth2_session_get.call_args.args[1] == 'https://api.tumblr.com/v2/user/info'
+
+
+def test_fetch_primary_blog_id_raises_exception(
+    tumblr_transfer_service, mock_oauth2_session_get_bad_response
+):
+    with pytest.raises(HTTPError):
+        tumblr_transfer_service.fetch_primary_blog_id()