skillbox/server/oauth/views.py

64 lines
1.8 KiB
Python

from authlib.integrations.base_client import OAuthError
from django.conf import settings
from django.shortcuts import redirect
from sentry_sdk import capture_exception
from oauth.hep_client import HepClient
from oauth.oauth_client import oauth
from oauth.models import OAuth2Token
from oauth.user_signup_login_handler import (
handle_user_and_verify_products,
EMAIL_NOT_VERIFIED,
UNKNOWN_ERROR,
)
from django.contrib.auth import login as dj_login
from core.logger import get_logger
logger = get_logger(__name__)
OAUTH_REDIRECT = "oauth-redirect"
def go_to_oauth(request, go_to_register=False):
hep_oauth_client = oauth.create_client("hep")
redirect_uri = settings.OAUTH_LOCAL_REDIRECT_URI
if go_to_register:
kwargs = {"redirect": "register"}
else:
kwargs = {}
return hep_oauth_client.authorize_redirect(request, redirect_uri, **kwargs)
def login(request):
return go_to_oauth(request)
def register(request):
return go_to_oauth(request, go_to_register=True)
def authorize(request):
hep_client = HepClient()
try:
token = oauth.hep.authorize_access_token(request)
user_data = hep_client.user_details(token_dict=token)
user, status_msg = handle_user_and_verify_products(user_data, token)
user.sync_with_hep_data(user_data)
except OAuthError as e:
logger.warning(f"OAuth error: {e}")
if not settings.DEBUG:
capture_exception(e)
return redirect(f"/{OAUTH_REDIRECT}?state={UNKNOWN_ERROR}")
if user and status_msg != EMAIL_NOT_VERIFIED:
dj_login(request, user)
OAuth2Token.objects.update_or_create_token(token, user)
if status_msg:
return redirect(f"/{OAUTH_REDIRECT}?state={status_msg}")
return redirect(f"/{OAUTH_REDIRECT}?state=success")