Skip to content
Snippets Groups Projects

Feature/comp 208 google o auth poc

Merged Bjarke Madsen requested to merge feature/COMP-208-Google-OAuth-POC into develop
3 files
+ 63
16
Compare changes
  • Side-by-side
  • Inline
Files
3
from flask_login import LoginManager, UserMixin # type: ignore
from flask import jsonify
from datetime import datetime
from compendium_v2.db import session_scope
from compendium_v2.db.auth_model import User
from flask_login import LoginManager # type: ignore
# TODO: implement user model as SQLAlchemy model
class User(UserMixin):
pass
def create_user(email: str, fullname: str, oidc_sub: str):
"""
Function used to create a new user in the database.
:param email: The email of the user
:param fullname: The full name of the user
:param oidc_sub: The OIDC subject identifier (ID) of the user
:return: The user object
"""
with session_scope() as session:
user = User(email=email, fullname=fullname, oidc_sub=oidc_sub)
session.add(user)
return user
def fetch_user(email: str):
def fetch_user(profile: dict):
"""
Function used to resolve an OIDC profile to a user in the database.
:param profile: OIDC profile information
:return: User object if the user exists, None otherwise.
"""
with session_scope() as session:
sub_id = profile['sub']
user = session.query(User).filter(User.oidc_sub == sub_id).first()
if user is None:
return None
user.last_login = datetime.utcnow()
return user
def user_loader(user_id: str):
"""
Function used to retrieve the internal user model for the user attempting login.
:param profile: The email of the user attempting login.
:param user_id: The ID of the user attempting login.
:return: User object if the user exists, None otherwise.
"""
# TODO: fetch user from database instead of just creating a user object
user = User()
user.id = email
return user
with session_scope() as session:
user = session.query(User).filter(User.id == user_id).first()
if user is None:
return None
user.last_login = datetime.utcnow()
return user
def unauth_handler():
return jsonify(success=False,
data={'login_required': True},
message='Authorize to access this page.'), 401
def setup_login_manager(login_manager: LoginManager):
login_manager.user_loader(fetch_user)
login_manager.user_loader(user_loader)
login_manager.unauthorized_handler(unauth_handler)
Loading