brewman/brewman/brewman/routers/auth/client.py

import uuid

import brewman.schemas.auth as schemas

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

from ...core.security import get_current_active_user as get_user
from ...db.session import SessionLocal
from ...models.auth import Client, LoginHistory
from ...schemas.auth import UserToken


router = APIRouter()


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


@router.put("/{id_}")
def update(
    id_: uuid.UUID,
    data: schemas.ClientIn,
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["clients"]),
):
    try:
        item: Client = db.query(Client).filter(Client.id == id_).first()
        item.enabled = data.enabled
        if item.enabled:
            item.otp = None
        item.name = data.name
        db.commit()
        return {}
    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.delete("/{id_}")
def delete(
    id_: uuid.UUID,
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["clients"]),
):
    try:
        item: Client = db.query(Client).filter(Client.id == id_).first()
        db.execute(LoginHistory.__table__.delete(LoginHistory.client_id == item.id))
        db.delete(item)
        db.commit()
        return {}
    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("/list")
async def show_list(
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["clients"]),
):
    list_ = db.query(Client).order_by(Client.name).all()
    clients = []
    for item in list_:
        last_login = (
            db.query(LoginHistory)
            .filter(LoginHistory.client_id == item.id)
            .order_by(desc(LoginHistory.date))
            .first()
        )
        last_login = (
            "Never"
            if last_login is None
            else last_login.date.strftime("%d-%b-%Y %H:%M")
        )
        clients.append(
            {
                "id": item.id,
                "code": item.code,
                "name": item.name,
                "enabled": item.enabled,
                "otp": item.otp,
                "creationDate": item.creation_date.strftime("%d-%b-%Y %H:%M"),
                "lastLogin": last_login,
            }
        )
    return clients


@router.get("/{id_}")
def show_id(
    id_: uuid.UUID,
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["clients"]),
):
    item: Client = db.query(Client).filter(Client.id == id_).first()
    return {
        "id": item.id,
        "code": item.code,
        "name": item.name,
        "enabled": item.enabled,
        "otp": item.otp,
    }