# UGE / L2 / Intro to relational databases / Python project prototype # Author: Pacien TRAN-GIRARD # Licence: EUPL-1.2 from fastapi import APIRouter, Depends, Request, Form, status from fastapi.responses import RedirectResponse, HTMLResponse from passlib.context import CryptContext import re from embrace.exceptions import IntegrityError from psycopg2.errors import UniqueViolation from app_sessions import UserSession, FlashMessageQueue from app_database import db_transaction from app_templating import TemplateRenderer # Password hashing context. # Handles proper salting and migration automatically. password_ctx = CryptContext(schemes=['bcrypt'], deprecated='auto') username_pattern = re.compile(r'^[a-zA-Z0-9-_]{4,16}$') to_homepage = RedirectResponse('/', status_code=status.HTTP_303_SEE_OTHER) to_wallet = RedirectResponse('/wallet', status_code=status.HTTP_303_SEE_OTHER) router = APIRouter() @router.get('/', response_class=HTMLResponse) def homepage( session: UserSession=Depends(UserSession), render: TemplateRenderer=Depends(TemplateRenderer), ): if session.is_logged_in(): return to_wallet return render('homepage.html.jinja') @router.post('/account/register') def account_register( session: UserSession=Depends(UserSession), messages: FlashMessageQueue=Depends(FlashMessageQueue), username: str=Form(...), password: str=Form(...), ): try: if username_pattern.match(username) is None: messages.add('error', 'Invalid username format.') return to_homepage if not 4 <= len(password) <= 32: messages.add('error', 'Invalid password length.') return to_homepage hash = password_ctx.hash(password) with db_transaction() as tx: user = tx.create_account(username=username, password_hash=hash) session.login(user.id) messages.add('success', 'Account succesfully created. Welcome!') return to_wallet except IntegrityError as exception: if isinstance(exception.__cause__, UniqueViolation): messages.add('error', 'This username is already taken.') return to_homepage else: raise exception @router.post('/account/login') def session_login( session: UserSession=Depends(UserSession), messages: FlashMessageQueue=Depends(FlashMessageQueue), username: str=Form(...), password: str=Form(...), ): with db_transaction() as tx: user = tx.fetch_account_username(username=username) if user is not None and password_ctx.verify(password, user.password_hash): session.login(user.id) messages.add('info', 'Welcome back!') return to_wallet else: messages.add('error', 'Invalid credentials.') return to_homepage @router.post('/account/logout') def session_logout( session: UserSession=Depends(UserSession), messages: FlashMessageQueue=Depends(FlashMessageQueue), ): if session.is_logged_in(): session.logout() messages.add('info', 'You have been successfully logged out.') return to_homepage