barker/barker/barker/routers/auth/user.py

import uuid
from typing import List, Optional

from fastapi import APIRouter, HTTPException, status, Depends, Security
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session

from ...schemas.auth import UserToken
from ...core.security import get_current_active_user as get_user
from ...db.session import SessionLocal
from barker.models.auth import User, Role
import barker.schemas.auth as schemas

router = APIRouter()


# Dependency
def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


@router.post("", response_model=schemas.User)
def save(
    data: schemas.UserIn, db: Session = Depends(get_db), user: UserToken = Security(get_user, scopes=["users"]),
):
    try:
        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:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e),
        )
    except Exception:
        db.rollback()
        raise


@router.get("/me", response_model=schemas.User)
def show_me(
    db: Session = Depends(get_db), user: UserToken = Depends(get_user),
):
    item = db.query(User).filter(User.id == user.id_).first()
    return user_info(item, db, user)


@router.put("/me", response_model=schemas.User)
def update_me(
    data: schemas.UserIn, db: Session = Depends(get_db), user: UserToken = Depends(get_user),
):
    try:
        item: User = db.query(User).filter(User.id == user.id_).first()
        if "users" in user.permissions:
            item.name = data.name
            item.locked_out = data.locked_out
            add_roles(item, data.roles, db)
        if data.password and item.password != data.password:
            item.password = data.password
        db.commit()
        return user_info(item, db, user)
    except SQLAlchemyError as e:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e),
        )
    except Exception:
        db.rollback()
        raise


@router.put("/{id_}", response_model=schemas.User)
def update(
    id_: uuid.UUID,
    data: schemas.UserIn,
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["users"]),
):
    try:
        item: User = db.query(User).filter(User.id == id_).first()
        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:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e),
        )
    except Exception:
        db.rollback()
        raise


def add_roles(user: User, roles: List[schemas.RoleItem], db: Session):
    for role in roles:
        ug = [g for g in user.roles if g.id == role.id_]
        ug = None if len(ug) == 0 else ug[0]
        if role.enabled and ug is None:
            user.roles.append(db.query(Role).filter(Role.id == role.id_).one())
        elif not role.enabled and ug:
            user.roles.remove(ug)


@router.delete("/{id_}")
def delete(
    id_: uuid.UUID, db: Session = Depends(get_db), user: UserToken = Security(get_user, scopes=["users"]),
):
    try:
        item: User = db.query(User).filter(User.id == id_).first()
        if item is None:
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="User not found",
            )
        else:
            raise HTTPException(
                status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="User deletion not implemented",
            )
    except Exception:
        db.rollback()
        raise


@router.get("")
def show_blank(
    db: Session = Depends(get_db), user: UserToken = Security(get_user, scopes=["users"]),
):
    return user_info(None, db, user)


@router.get("/list", response_model=List[schemas.UserList])
async def show_list(
    db: Session = Depends(get_db), user: UserToken = Security(get_user, scopes=["users"]),
):
    return [
        {
            "id": item.id,
            "name": item.name,
            "lockedOut": item.locked_out,
            "roles": [p.name for p in sorted(item.roles, key=lambda p: p.name)],
        }
        for item in db.query(User).order_by(User.name).all()
    ]


@router.get("/active")
async def show_active(db: Session = Depends(get_db), user: UserToken = Depends(get_user)):
    return [{"name": item.name} for item in db.query(User).filter(User.locked_out == False).order_by(User.name)]


@router.get("/{id_}", response_model=schemas.User)
def show_id(
    id_: uuid.UUID, db: Session = Depends(get_db), user: UserToken = Security(get_user, scopes=["users"]),
):
    item = db.query(User).filter(User.id == id_).first()
    return user_info(item, db, user)


def user_info(item: Optional[User], db: Session, user: UserToken):
    if item is None:
        return {
            "name": "",
            "lockedOut": False,
            "roles": [{"id": r.id, "name": r.name, "enabled": False} for r in db.query(Role).order_by(Role.name).all()],
        }
    else:
        return {
            "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.query(Role).order_by(Role.name).all()
            ]
            if "users" in user.permissions
            else [],
        }