refactor: create base class OAuthBackend

Author: joonas-yoon

fastapi/app/auth/base/libs.py

@@ -0,0 +1,47 @@
+from datetime import datetime
+from typing import Any, Optional
+
+from fastapi import Response
+from fastapi_users import models
+from fastapi_users.authentication import AuthenticationBackend
+from fastapi_users.authentication.strategy import Strategy
+from fastapi_users.authentication.transport import Transport
+from fastapi_users.types import DependencyCallable
+
+from ..models import User
+
+
+class OAuthBackend(AuthenticationBackend):
+    def __init__(
+        self,
+        name: str,
+        transport: Transport,
+        get_strategy: DependencyCallable[Strategy[models.UP, models.ID]],
+        oauth_name: str,
+        has_profile_callback: bool = False,
+    ):
+        super().__init__(name, transport, get_strategy)
+        self.oauth_name = oauth_name
+        self.has_profile_callback = has_profile_callback
+
+    async def login(self, strategy: Strategy, user: User, response: Response) -> Any:
+        strategy_response = await super().login(strategy, user, response)
+        user.last_login_at = datetime.now()
+        if self.has_profile_callback:
+            token = self.get_access_token(user)
+            profile = self.get_profile(token)
+            user = self.update_profile(user, profile)
+        await user.save()
+        return strategy_response
+
+    def get_access_token(self, user: User) -> Optional[str]:
+        for account in user.oauth_accounts:
+            if account.oauth_name == self.oauth_name:
+                return account.access_token
+        return None
+
+    def get_profile(self, access_token: str) -> dict:
+        return {}
+
+    def update_profile(self, user: User, profile: dict) -> User:
+        return user

fastapi/app/auth/github/libs.py

@@ -1,64 +1,40 @@
-from datetime import datetime
-from typing import Any, Optional
-
 import requests
-from fastapi import Response
-from fastapi_users.authentication import AuthenticationBackend
-from fastapi_users.authentication.strategy import Strategy
 
+from ..base.libs import OAuthBackend
 from ..exceptions import BadCredentialException
 from ..libs import bearer_transport, get_jwt_strategy
 from ..models import User
 from .constants import GITHUB_USERINFO_URL
 
 
-class GithubAuthBackend(AuthenticationBackend):
-    async def login(self, strategy: Strategy, user: User, response: Response) -> Any:
-        strategy_response = await super().login(strategy, user, response)
-        token = self.get_google_access_token(user)
-        profile = get_profile(token)
-        await update_profile(user, profile).save()
-        await user.save()
-        return strategy_response
-
-    def get_google_access_token(self, user: User) -> Optional[str]:
-        for account in user.oauth_accounts:
-            if account.oauth_name == 'github':
-                return account.access_token
-        return None
-
-
-def get_profile(access_token: str) -> dict:
-    headers = dict(Authorization=f'Bearer {access_token}')
-    response = requests.get(GITHUB_USERINFO_URL, headers=headers)
-    if not response.ok:
-        raise BadCredentialException(
-            'Failed to get user information from Github.')
-    profile = dict(response.json())
-    name = profile.get('name').split()
-    try:
-        profile.update({
-            "first_name": name[0],
-            "last_name": name[1],
-        })
-    except:
-        pass
-    return profile
-
-
-def update_profile(user: User, profile: dict) -> User:
-    if user.first_name == None:
-        user.first_name = profile.get('first_name')
-    if user.last_name == None:
-        user.last_name = profile.get('last_name')
-    if user.picture == None:
-        user.picture = profile.get('avatar_url')
-    user.last_login_at = datetime.now()
-    return user
-
-
-auth_backend_github = GithubAuthBackend(
+class GithubAuthBackend(OAuthBackend):
+    def get_profile(self, access_token: str) -> dict:
+        headers = dict(Authorization=f'Bearer {access_token}')
+        response = requests.get(GITHUB_USERINFO_URL, headers=headers)
+        if not response.ok:
+            raise BadCredentialException(
+                'Failed to get user information from Github.')
+        return response.json()
+
+    def update_profile(self, user: User, profile: dict) -> User:
+        user = super().update_profile(user, profile)
+        try:
+            name = profile.get('name').split()
+            if user.first_name == None:
+                user.first_name = name[0]
+            if user.last_name == None:
+                user.last_name = name[1]
+        except:
+            pass
+        if user.picture == None:
+            user.picture = profile.get('avatar_url')
+        return user
+
+
+auth_backend = GithubAuthBackend(
     name="jwt-github",
     transport=bearer_transport,
     get_strategy=get_jwt_strategy,
+    oauth_name='github',
+    has_profile_callback=True,
 )

fastapi/app/auth/github/routes.py

@@ -3,7 +3,7 @@
 from httpx_oauth.clients.github import GitHubOAuth2
 
 from ..libs import fastapi_users
-from .libs import auth_backend_github
+from .libs import auth_backend
 
 github_oauth_client = GitHubOAuth2(
     client_id=Configs.GITHUB_CLIENT_ID,
@@ -12,7 +12,7 @@
 
 github_oauth_router = fastapi_users.get_oauth_router(
     oauth_client=github_oauth_client,
-    backend=auth_backend_github,
+    backend=auth_backend,
     state_secret=Configs.SECRET_KEY,
     redirect_url=Configs.GITHUB_CALLBACK_URL,
     associate_by_email=True,

fastapi/app/auth/google/libs.py

@@ -1,54 +1,36 @@
-from datetime import datetime
-from typing import Any, Optional
-
 import requests
-from fastapi import Response
-from fastapi_users.authentication import AuthenticationBackend
-from fastapi_users.authentication.strategy import Strategy
 
+from ..base.libs import OAuthBackend
 from ..exceptions import BadCredentialException
 from ..libs import bearer_transport, get_jwt_strategy
 from ..models import User
 from .constants import GOOGLE_USERINFO_URL
 
 
-class GoogleAuthBackend(AuthenticationBackend):
-    async def login(self, strategy: Strategy, user: User, response: Response) -> Any:
-        strategy_response = await super().login(strategy, user, response)
-        token = self.get_google_access_token(user)
-        profile = get_profile(token)
-        await update_profile(user, profile).save()
-        return strategy_response
-
-    def get_google_access_token(self, user: User) -> Optional[str]:
-        for account in user.oauth_accounts:
-            if account.oauth_name == 'google':
-                return account.access_token
-        return None
-
-
-def get_profile(access_token: str) -> dict:
-    response = requests.get(url=GOOGLE_USERINFO_URL,
-                            params={'access_token': access_token})
-    if not response.ok:
-        raise BadCredentialException(
-            'Failed to get user information from Google.')
-    return response.json()
-
+class GoogleAuthBackend(OAuthBackend):
+    def get_profile(self, access_token: str) -> dict:
+        response = requests.get(url=GOOGLE_USERINFO_URL,
+                                params={'access_token': access_token})
+        if not response.ok:
+            raise BadCredentialException(
+                'Failed to get user information from Google.')
+        return response.json()
 
-def update_profile(user: User, profile: dict) -> User:
-    if user.first_name == None:
-        user.first_name = profile.get('given_name')
-    if user.last_name == None:
-        user.last_name = profile.get('family_name')
-    if user.picture == None:
-        user.picture = profile.get('picture')
-    user.last_login_at = datetime.now()
-    return user
+    def update_profile(self, user: User, profile: dict) -> User:
+        user = super().update_profile(user, profile)
+        if user.first_name == None:
+            user.first_name = profile.get('given_name')
+        if user.last_name == None:
+            user.last_name = profile.get('family_name')
+        if user.picture == None:
+            user.picture = profile.get('picture')
+        return user
 
 
-auth_backend_google = GoogleAuthBackend(
+auth_backend = GoogleAuthBackend(
     name="jwt-google",
     transport=bearer_transport,
     get_strategy=get_jwt_strategy,
+    oauth_name='google',
+    has_profile_callback=True,
 )

fastapi/app/auth/google/routes.py

@@ -4,7 +4,7 @@
 
 from ..libs import fastapi_users
 from .constants import GOOGLE_SCOPE_EMAIL, GOOGLE_SCOPE_PROFILE
-from .libs import auth_backend_google
+from .libs import auth_backend
 
 google_oauth_client = GoogleOAuth2(
     client_id=Configs.GOOGLE_CLIENT_ID,
@@ -16,7 +16,7 @@
 
 google_oauth_router = fastapi_users.get_oauth_router(
     oauth_client=google_oauth_client,
-    backend=auth_backend_google,
+    backend=auth_backend,
     state_secret=Configs.SECRET_KEY,
     redirect_url=Configs.GOOGLE_CALLBACK_URL,
     associate_by_email=True,

fastapi/app/auth/kakao/libs.py

@@ -1,56 +1,39 @@
-from datetime import datetime
 import json
-from typing import Any, Optional
 
 import requests
-from fastapi import Response
-from fastapi_users.authentication import AuthenticationBackend
-from fastapi_users.authentication.strategy import Strategy
 
+from ..base.libs import OAuthBackend
 from ..exceptions import BadCredentialException
 from ..libs import bearer_transport, get_jwt_strategy
 from ..models import User
 from .constants import KAKAO_USERINFO_URL
 
 
-class KakaoAuthBackend(AuthenticationBackend):
-    async def login(self, strategy: Strategy, user: User, response: Response) -> Any:
-        strategy_response = await super().login(strategy, user, response)
-        token = self.get_access_token(user)
-        profile = get_profile(token)
-        await update_profile(user, profile).save()
-        return strategy_response
+class KakaoAuthBackend(OAuthBackend):
+    def get_profile(self, access_token: str) -> dict:
+        headers = dict(Authorization=f'Bearer {access_token}')
+        response = requests.post(KAKAO_USERINFO_URL,
+                                 headers=headers,
+                                 params={"property_keys": json.dumps(["kakao_account.profile"])})
+        if not response.ok:
+            raise BadCredentialException(
+                'Failed to get user information from Kakao.')
 
-    def get_access_token(self, user: User) -> Optional[str]:
-        for account in user.oauth_accounts:
-            if account.oauth_name == 'kakao':
-                return account.access_token
-        return None
+        return response.json()
 
+    def update_profile(self, user: User, profile: dict) -> User:
+        user = super().update_profile(user, profile)
+        kakao_account = profile.get('kakao_account')
+        profile = kakao_account.get('profile')
+        if user.picture == None:
+            user.picture = profile.get('profile_image_url')
+        return user
 
-def get_profile(access_token: str) -> dict:
-    headers = dict(Authorization=f'Bearer {access_token}')
-    response = requests.post(KAKAO_USERINFO_URL,
-                             headers=headers,
-                             params={"property_keys": json.dumps(["kakao_account.profile"])})
-    if not response.ok:
-        raise BadCredentialException(
-            'Failed to get user information from Kakao.')
 
-    profile = dict(response.json())
-    kakao_account = profile.get('kakao_account')
-    return kakao_account.get('profile')
-
-
-def update_profile(user: User, profile: dict) -> User:
-    if user.picture == None:
-        user.picture = profile.get('profile_image_url')
-    user.last_login_at = datetime.now()
-    return user
-
-
-auth_backend_kakao = KakaoAuthBackend(
+auth_backend = KakaoAuthBackend(
     name="jwt-kakao",
     transport=bearer_transport,
     get_strategy=get_jwt_strategy,
+    oauth_name='kakao',
+    has_profile_callback=True,
 )

fastapi/app/auth/kakao/routes.py

@@ -1,9 +1,8 @@
-
 from app.configs import Configs
 
 from ..libs import fastapi_users
 from .client import KakaoOAuth2
-from .libs import auth_backend_kakao
+from .libs import auth_backend
 
 kakao_oauth_client = KakaoOAuth2(
     client_id=Configs.KAKAO_CLIENT_ID,
@@ -15,7 +14,7 @@
 
 kakao_oauth_router = fastapi_users.get_oauth_router(
     oauth_client=kakao_oauth_client,
-    backend=auth_backend_kakao,
+    backend=auth_backend,
     state_secret=Configs.SECRET_KEY,
     redirect_url=Configs.KAKAO_CALLBACK_URL,
     associate_by_email=True,