Coverage for src / mysingle / auth / oauth_manager.py: 0%

1# path: app/auth/providers.py 

2 

3import logging 

4import secrets 

5from typing import Optional, Union 

6 

7import httpx 

8from httpx_oauth.clients.google import GoogleOAuth2 

9from httpx_oauth.clients.kakao import KakaoOAuth2 

10from httpx_oauth.clients.naver import NaverOAuth2 

11 

12from ..core.config import settings 

13from .schemas.oauth2 import ( 

14 BaseOAuthToken, 

15 GoogleProfile, 

16 GoogleToken, 

17 KakaoProfile, 

18 KakaoToken, 

19 NaverProfile, 

20 NaverToken, 

21) 

22 

23logger = logging.getLogger(__name__) 

24 

25# --------------------------- 

26# OAuth2 Clients 

27# --------------------------- 

28google_client = GoogleOAuth2( 

29 client_id=settings.GOOGLE_CLIENT_ID, 

30 client_secret=settings.GOOGLE_CLIENT_SECRET, 

31) 

32kakao_client = KakaoOAuth2( 

33 client_id=settings.KAKAO_CLIENT_ID, 

34 client_secret=settings.KAKAO_CLIENT_SECRET, 

35) 

36naver_client = NaverOAuth2( 

37 client_id=settings.NAVER_CLIENT_ID, 

38 client_secret=settings.NAVER_CLIENT_SECRET, 

39) 

40 

41AVAILABLE_PROVIDERS = { 

42 "google": google_client, 

43 "kakao": kakao_client, 

44 "naver": naver_client, 

45} 

46 

47 

48class OAuthManager: 

49 """ 

50 OAuthManager 클래스를 통해 

51 - Provider별 authorize URL 

52 - Callback 시 Access Token & Profile 

53 - User DB 업데이트 

54 를 일관되게 처리. 

55 """ 

56 

57 @staticmethod 

58 def get_provider_client(provider: str): 

59 if provider not in AVAILABLE_PROVIDERS: 

60 raise ValueError(f"Unknown OAuth provider: {provider}") 

61 return AVAILABLE_PROVIDERS[provider] 

62 

63 @staticmethod 

64 def get_redirect_uri(provider: str) -> str: 

65 return f"{settings.FRONTEND_URL}/api/auth/oauth2/{provider}/callback" 

66 

67 @staticmethod 

68 async def generate_auth_url( 

69 provider: str, state: Optional[str], redirect_uri: str | None = None 

70 ) -> str: 

71 client = OAuthManager.get_provider_client(provider) 

72 if not redirect_uri: 

73 redirect_uri = OAuthManager.get_redirect_uri(provider) 

74 state_string = state or secrets.token_urlsafe(16) 

75 scope = None 

76 if provider == "google": 

77 scope = ["openid", "email", "profile"] 

78 try: 

79 authorization_url = await client.get_authorization_url( 

80 redirect_uri=redirect_uri, state=state_string, scope=scope 

81 ) 

82 return str(authorization_url) 

83 except Exception as e: 

84 logger.error(f"{provider} Authorization URL 생성 오류: {e}") 

85 raise ValueError(f"{provider.capitalize()} Authorization URL 생성 실패") 

86 

87 @staticmethod 

88 async def get_access_token_and_profile( 

89 provider: str, 

90 code: str, 

91 redirect_uri: str, 

92 ) -> tuple[BaseOAuthToken, Union[GoogleProfile, KakaoProfile, NaverProfile]]: 

93 """ 

94 1) Access Token 획득 

95 2) Provider별 Profile API 호출 

96 3) (token_data, profile_data) 반환 

97 """ 

98 client = OAuthManager.get_provider_client(provider) 

99 try: 

100 token_response = await client.get_access_token( 

101 code=code, 

102 redirect_uri=redirect_uri, 

103 ) 

104 except Exception as e: 

105 logger.error(f"{provider} 토큰 획득 오류: {e}") 

106 raise ValueError(f"{provider.capitalize()} 토큰 획득 실패") 

107 token_data: BaseOAuthToken 

108 profile_data: Union[GoogleProfile, KakaoProfile, NaverProfile] 

109 async with httpx.AsyncClient() as httpc: 

110 if provider == "google": 

111 token_data = GoogleToken(**token_response) 

112 res = await httpc.get( 

113 "https://www.googleapis.com/oauth2/v2/userinfo", 

114 params={"access_token": token_data.access_token}, 

115 ) 

116 res.raise_for_status() 

117 raw_profile = res.json() 

118 profile_data = GoogleProfile(**raw_profile) 

119 elif provider == "kakao": 

120 token_data = KakaoToken(**token_response) 

121 res = await httpc.get( 

122 "https://kapi.kakao.com/v2/user/me", 

123 headers={"Authorization": f"Bearer {token_data.access_token}"}, 

124 ) 

125 res.raise_for_status() 

126 raw_profile = res.json() 

127 profile_data = KakaoProfile(**raw_profile) 

128 elif provider == "naver": 

129 token_data = NaverToken(**token_response) 

130 res = await httpc.get( 

131 "https://openapi.naver.com/v1/nid/me", 

132 headers={"Authorization": f"Bearer {token_data.access_token}"}, 

133 ) 

134 res.raise_for_status() 

135 raw_profile = res.json() 

136 profile_data = NaverProfile(**raw_profile["response"]) 

137 else: 

138 raise ValueError(f"Not implemented provider: {provider}") 

139 return token_data, profile_data 

140 

141 

142oauth_manager = OAuthManager()