barker/barker/barker/routers/user.py

import uuid

import barker.schemas.user as schemas

from barker.core.security import get_current_active_user as get_user
from barker.db.session import SessionFuture
from barker.models.kot import Kot
from barker.models.login_history import LoginHistory
from barker.models.reprint import Reprint
from barker.models.role import Role
from barker.models.user import User
from barker.models.user_roles import user_roles
from barker.models.voucher import Voucher
from barker.schemas.user_token import UserToken
from fastapi import APIRouter, Depends, HTTPException, Security, status
from sqlalchemy import delete, select, update
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session


router = APIRouter()


@router.post("", response_model=schemas.User)
def save(
    data: schemas.UserIn,
    user: UserToken = Security(get_user, scopes=["users"]),
) -> schemas.User:
    try:
        with SessionFuture() as db:
            item = User(name=data.name, password=data.password, locked_out=data.locked_out)
            db.add(item)
            add_roles(item, data.roles, db)
            db.commit()
            return user_info(item, db, user)
    except SQLAlchemyError as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e),
        )


@router.get("/me", response_model=schemas.UserMe)
def show_me(
    user: UserToken = Depends(get_user),
) -> schemas.UserMe:
    with SessionFuture() as db:
        item = db.execute(select(User).where(User.id == user.id_)).scalar_one()
        return user_me(item)


@router.put("/me", response_model=schemas.UserMe)
def update_me(
    data: schemas.UserIn,
    user: UserToken = Depends(get_user),
) -> schemas.UserMe:
    try:
        with SessionFuture() as db:
            item: User = db.execute(select(User).where(User.id == user.id_)).scalar_one()
            if data.password and item.password != data.password:
                item.password = data.password
            db.commit()
            return user_me(item)
    except SQLAlchemyError as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e),
        )


@router.put("/{id_}", response_model=schemas.User)
def update_route(
    id_: uuid.UUID,
    data: schemas.UserIn,
    user: UserToken = Security(get_user, scopes=["users"]),
) -> schemas.User:
    try:
        with SessionFuture() as db:
            item: User = db.execute(select(User).where(User.id == id_)).scalar_one()
            item.name = data.name
            if data.password and item.password != data.password:
                item.password = data.password
            item.locked_out = data.locked_out
            add_roles(item, data.roles, db)
            db.commit()
            return user_info(item, db, user)
    except SQLAlchemyError as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e),
        )


def add_roles(user: User, roles: list[schemas.RoleItem], db: Session):
    for role in roles:
        ug = next((g for g in user.roles if g.id == role.id_), None)
        if role.enabled and ug is None:
            user.roles.append(db.execute(select(Role).where(Role.id == role.id_)).scalar_one())
        elif not role.enabled and ug:
            user.roles.remove(ug)


@router.delete("/{id_}")
def delete_route(
    id_: uuid.UUID,
    user: UserToken = Security(get_user, scopes=["users"]),
):
    if id_ == user.id_:
        raise HTTPException(
            status_code=status.HTTP_501_NOT_IMPLEMENTED,
            detail="You cannot delete yourself",
        )
    with SessionFuture() as db:
        item: User = db.execute(select(User).where(User.id == id_)).scalar_one()
        admin: User = db.execute(select(User).where(User.name.ilike("admin"))).scalar_one()
        db.execute(update(Voucher).where(Voucher.user_id == item.id).values(user_id=admin.id))
        db.execute(update(Kot).where(Kot.user_id == item.id).values(user_id=admin.id))
        db.execute(update(Reprint).where(Reprint.user_id == item.id).values(user_id=admin.id))
        db.execute(delete(LoginHistory).where(LoginHistory.user_id == item.id))
        db.execute(delete(user_roles).where(user_roles.c.user_id == item.id))
        db.execute(delete(User).where(User.id == item.id))
        db.commit()


@router.get("", response_model=schemas.UserIn)
def show_blank(
    user: UserToken = Security(get_user, scopes=["users"]),
) -> schemas.UserIn:
    with SessionFuture() as db:
        return blank_user_info(db)


@router.get("/list", response_model=list[schemas.UserList])
def show_list(
    user: UserToken = Security(get_user, scopes=["users"]),
) -> list[schemas.UserList]:
    with SessionFuture() as db:
        return [
            schemas.UserList(
                id=item.id,
                name=item.name,
                lockedOut=item.locked_out,
                roles=[p.name for p in sorted(item.roles, key=lambda p: p.name)],
                lastDevice=item.login_history[0].device.name if len(item.login_history) else "Never",
                lastDate=item.login_history[0].date if len(item.login_history) else None,
            )
            for item in db.execute(select(User).order_by(User.name)).scalars().all()
        ]


@router.get("/{id_}", response_model=schemas.User)
def show_id(
    id_: uuid.UUID,
    user: UserToken = Security(get_user, scopes=["users"]),
) -> schemas.User:
    with SessionFuture() as db:
        item = db.execute(select(User).where(User.id == id_)).scalar_one()
        return user_info(item, db, user)


def user_info(item: User, db: Session, user: UserToken) -> schemas.User:
    return schemas.User(
        id=item.id,
        name=item.name,
        password="",
        lockedOut=item.locked_out,
        roles=[
            {
                "id": r.id,
                "name": r.name,
                "enabled": True if r in item.roles else False,
            }
            for r in db.execute(select(Role).order_by(Role.name)).scalars().all()
        ]
        if "users" in user.permissions
        else [],
    )


def user_me(item: User) -> schemas.UserMe:
    return schemas.UserMe(
        id="me",
        name=item.name,
        password="",
        lockedOut=item.locked_out,
        roles=[],
    )


def blank_user_info(db: Session) -> schemas.UserIn:
    return schemas.UserIn(
        name="",
        password="",
        lockedOut=False,
        roles=[
            {"id": r.id, "name": r.name, "enabled": False}
            for r in db.execute(select(Role).order_by(Role.name)).scalars().all()
        ],
    )