48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
from schemas import LoginSchema
|
|
from repositories import UserRepository
|
|
from mapper import UserMapper
|
|
from google.oauth2 import id_token
|
|
from google.auth.transport import requests
|
|
from configs import Config
|
|
from exception import AuthException
|
|
from werkzeug.security import check_password_hash
|
|
|
|
|
|
class AuthService:
|
|
def __init__(self, userRepository: UserRepository):
|
|
self.user_repository = userRepository
|
|
|
|
def verify_google_id_token(self, id_token_str):
|
|
payload = id_token.verify_oauth2_token(
|
|
id_token_str, requests.Request(), Config.GOOGLE_CLIENT_ID
|
|
)
|
|
|
|
if not payload:
|
|
raise AuthException("Invalid Google ID Token")
|
|
|
|
google_id = payload.get("sub")
|
|
email = payload.get("email")
|
|
|
|
existing_user = self.user_repository.get_by_google_id(google_id)
|
|
if existing_user:
|
|
if existing_user.email == email:
|
|
return existing_user
|
|
raise AuthException("Email not match")
|
|
|
|
new_user = UserMapper.from_google_payload(google_id, email, payload)
|
|
|
|
user_id = self.user_repository.insert_user(user_data=new_user)
|
|
|
|
return self.user_repository.get_user_by_id(user_id=user_id)
|
|
|
|
def login(self, data: LoginSchema):
|
|
user_data = self.user_repository.get_user_by_email(data.email)
|
|
|
|
if user_data is None:
|
|
return None
|
|
|
|
if check_password_hash(user_data.password, data.password):
|
|
user_data.password = None
|
|
return user_data
|
|
return None
|