diff --git a/src/pardner/services/tumblr.py b/src/pardner/services/tumblr.py index 9dad6b5..4955005 100644 --- a/src/pardner/services/tumblr.py +++ b/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] = {}, diff --git a/tests/test_transfer_services/test_tumblr.py b/tests/test_transfer_services/test_tumblr.py index 05c9c9f..b3e27e6 100644 --- a/tests/test_transfer_services/test_tumblr.py +++ b/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()