import re import peewee from quart import session, redirect, url_for, current_app from quart_keycloak import Keycloak, KeycloakAuthToken, KeycloakLogoutRequest from yellow.factory import keycloak from yellow.models import User @keycloak.after_login() async def handle_user_login(auth_token: KeycloakAuthToken): username = auth_token.username uid = auth_token.sub if not re.match(r"^[a-zA-Z0-9_\.-]+$", username): raise Exception("bad username") try: user = User.select().where(User.username == username).get() except Exception as ex: user = None if not user: # create new user if it does not exist yet current_app.logger.info(f'User {username} not found, creating') try: user = User.create(id=uid, username=username) except Exception as ex: current_app.logger.debug(f'User {username}, creation error') raise # user is now logged in session['user'] = user.to_json() return redirect(url_for('bp_routes.dashboard'))