kers/oauth/views.py

import logging

import requests
from django.conf import settings
from django.contrib.auth import login
from django.http.request import HttpRequest
from django.shortcuts import redirect

from users.models import CustomUser

logger = logging.getLogger(__file__)


class OAuthException(Exception):
    pass


def register(_):
    RESPONSE_TYPE = 'code'
    return redirect(f'{settings.OAUTH["AUTHORIZE_URI"]}?'
                    f'response_type={RESPONSE_TYPE}&'
                    f'client_id={settings.OAUTH["CLIENT_ID"]}&'
                    f'redirect_uri={settings.OAUTH["REDIRECT_URI"]}')


def register_callback(req: HttpRequest):
    code = req.GET['code']
    response = requests.post(settings.OAUTH["AUTHORIZE_URI"],
                             data={'code': code,
                                   'grant_type': 'authorization_code',
                                   'client_id': settings.OAUTH["CLIENT_ID"],
                                   'client_secret': settings.OAUTH["CLIENT_SECRET"],
                                   'redirect_uri': settings.OAUTH["REDIRECT_URI"]})
    try:
        if response.status_code == 200:
            json: dict = response.json()
            # TODO: maybe later do something with the refresh token.
            user: dict = user_info(json['access_token'])
            if 'username' not in user.keys() or 'id' not in user.keys():
                raise OAuthException(f'username and id are expected values: {user}')
            else:
                logger.debug(f'Succesfully authenticated user: {user["username"]} with id: {user["id"]}')
                validated_user = validate_user(user['id'], user['username'])
                login(req, validated_user)
                redirect('/')
        else:
            raise OAuthException(f'Status code not 200, response: {response.json()}')
    except OAuthException as e:
        logger.error(e)

    return register('')


def validate_user(zeus_id, username) -> CustomUser:
    user = CustomUser.objects.get(zeus_id=zeus_id)
    if user is None:
        return CustomUser.objects.create_user(zeus_id, username)
    user.username = username
    user.save()
    return user


def user_info(access_token):
    r = requests.get(settings.OAUTH["USER_API_URI"], headers={'Authorization': f'Bearer {access_token}'})
    return r.json()