dashboard/helpers/hydra_oauth.py

from flask import request, session
from requests_oauthlib import OAuth2Session

from config import *
from helpers import HydraError


class HydraOauth:
    @staticmethod
    def authorize():
        try:
            hydra = OAuth2Session(HYDRA_CLIENT_ID)
            authorization_url, state = hydra.authorization_url(
                HYDRA_AUTHORIZATION_BASE_URL
            )

            return authorization_url
        except Exception as err:
            raise HydraError(str(err), 500)

    @staticmethod
    def get_token(state, code):
        try:
            hydra = OAuth2Session(
                client_id=HYDRA_CLIENT_ID,
                state=state,
            )
            token = hydra.fetch_token(
                token_url=TOKEN_URL,
                code=code,
                client_secret=HYDRA_CLIENT_SECRET,
                include_client_id=True,
            )

            session["hydra_token"] = token
            return token
        except Exception as err:
            raise HydraError(str(err), 500)

    @staticmethod
    def get_user_info():
        try:
            hydra = OAuth2Session(
                client_id=HYDRA_CLIENT_ID, token=session["hydra_token"]
            )
            user_info = hydra.get("{}/userinfo".format(HYDRA_PUBLIC_URL))

            return user_info.json()
        except Exception as err:
            raise HydraError(str(err), 500)