Changeset View
Changeset View
Standalone View
Standalone View
profiles/blender_id.py
| """Implement integration with Blender ID: handling of OAuth2 session, fetching user info etc.""" | """Implement integration with Blender ID: handling of OAuth2 session, fetching user info etc.""" | ||||
| from typing import Dict, Any | from typing import Dict, Any, Tuple | ||||
| from requests_oauthlib import OAuth2Session | from requests_oauthlib import OAuth2Session | ||||
| from urllib.parse import urljoin | from urllib.parse import urljoin, urlparse | ||||
| import logging | import logging | ||||
| import io | |||||
| import pathlib | |||||
| from blender_id_oauth_client.models import OAuthUserInfo, OAuthToken | from blender_id_oauth_client.models import OAuthUserInfo, OAuthToken | ||||
| from blender_id_oauth_client.views import blender_id_oauth_settings | from blender_id_oauth_client.views import blender_id_oauth_settings | ||||
| logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
| class BIDSession: | class BIDSession: | ||||
| """Wrap up interactions with Blender ID, such as fetching user info and avatar.""" | """Wrap up interactions with Blender ID, such as fetching user info and avatar.""" | ||||
| _anonymous_session = None | |||||
| def __init__(self): | def __init__(self): | ||||
| """Initialise Blender ID client settings.""" | """Initialise Blender ID client settings.""" | ||||
| self.settings = blender_id_oauth_settings() | self.settings = blender_id_oauth_settings() | ||||
| def _make_session(self, access_token: str = None) -> OAuth2Session: | def _make_session(self, access_token: str = None) -> OAuth2Session: | ||||
| """Return a new OAuth2 session, optionally authenticated with an access token.""" | """Return a new OAuth2 session, optionally authenticated with an access token.""" | ||||
| if access_token: | if access_token: | ||||
| return OAuth2Session(self.settings.client, token={'access_token': access_token,},) | return OAuth2Session(self.settings.client, token={'access_token': access_token}) | ||||
| return OAuth2Session(self.settings.client) | return OAuth2Session(self.settings.client) | ||||
| @property | |||||
| def session(self): | |||||
| """ | |||||
| Return a reusable "anonymous" OAuth2Session for fetching avatars from Blender ID. | |||||
| Create it the first time this property is accessed. | |||||
| """ | |||||
| if not self._anonymous_session: | |||||
| self._anonymous_session = self._make_session() | |||||
| return self._anonymous_session | |||||
| @classmethod | @classmethod | ||||
| def get_oauth_user_info(cls, oauth_user_id: str) -> OAuthUserInfo: | def get_oauth_user_info(cls, oauth_user_id: str) -> OAuthUserInfo: | ||||
| """Return OAuthUserInfo record for a given Blender ID. | """Return OAuthUserInfo record for a given Blender ID. | ||||
| Used primarily to look up our own user ID associated with an external Blender ID, | Used primarily to look up our own user ID associated with an external Blender ID, | ||||
| for example in the user modified webhook. | for example in the user modified webhook. | ||||
| """ | """ | ||||
| return ( | return ( | ||||
| Show All 28 Lines | def get_user_info(self, oauth_user_id: str) -> Dict[str, Any]: | ||||
| payload = resp.json() | payload = resp.json() | ||||
| assert isinstance(payload, dict) | assert isinstance(payload, dict) | ||||
| return payload | return payload | ||||
| def get_avatar_url(self, oauth_user_id: str) -> str: | def get_avatar_url(self, oauth_user_id: str) -> str: | ||||
| """Return a Blender ID URL to the avatar for a given OAuth ID.""" | """Return a Blender ID URL to the avatar for a given OAuth ID.""" | ||||
| return urljoin(self.settings.url_base, f'api/user/{oauth_user_id}/avatar') | return urljoin(self.settings.url_base, f'api/user/{oauth_user_id}/avatar') | ||||
| def get_avatar(self, oauth_user_id: str) -> Tuple[str, io.BytesIO]: | |||||
| """Retrieve an avatar from Blender ID service using an OAuth2 session. | |||||
| Return file name and content of an avatar for the given 'oauth_user_id'. | |||||
| """ | |||||
| resp = self.session.get(self.get_avatar_url(oauth_user_id)) | |||||
| resp.raise_for_status() | |||||
| name = pathlib.Path(urlparse(resp.url).path).name | |||||
| return name, io.BytesIO(resp.content) | |||||