139 lines
4.2 KiB
Python
139 lines
4.2 KiB
Python
import logging
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlmodel import Session, select
|
|
|
|
from app.model.models import (
|
|
AccessAuthorizationCreate,
|
|
AccessAuthorizationDB,
|
|
AccessAuthorizationResponse,
|
|
AccessAuthorizationUpdate,
|
|
GroupDB,
|
|
GroupResponse,
|
|
OneShotAccess,
|
|
)
|
|
from app.services.auth import auth_is_admin
|
|
from app.services.database import add_and_refresh, get_session
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
aa_router = APIRouter(prefix="/api/v1/aa", tags=["AccessAuth"])
|
|
|
|
|
|
@aa_router.post("/", response_model=AccessAuthorizationResponse)
|
|
def add_accessauth(
|
|
*,
|
|
db: Session = Depends(get_session),
|
|
aa: AccessAuthorizationCreate,
|
|
admin: bool = Depends(auth_is_admin),
|
|
):
|
|
logger.info(f"Creating accessauth with data: {aa}")
|
|
if aa.oneshot is not None:
|
|
oneshot = OneShotAccess.model_validate(aa.oneshot)
|
|
else:
|
|
oneshot = None
|
|
db_aa = AccessAuthorizationDB(
|
|
name=aa.name,
|
|
type=aa.type,
|
|
is_active=aa.is_active,
|
|
oneshot=oneshot,
|
|
)
|
|
return add_and_refresh(db, db_aa)
|
|
|
|
|
|
@aa_router.get("/", response_model=list[AccessAuthorizationResponse])
|
|
def get_all_accessauths(
|
|
db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
|
|
):
|
|
return db.exec(select(AccessAuthorizationDB)).all()
|
|
|
|
|
|
@aa_router.get("/{aa_id}", response_model=AccessAuthorizationResponse)
|
|
def get_one_accessauth(
|
|
*,
|
|
db: Session = Depends(get_session),
|
|
aa_id: int,
|
|
admin: bool = Depends(auth_is_admin),
|
|
):
|
|
db_aa = db.get(AccessAuthorizationDB, aa_id)
|
|
if db_aa is None:
|
|
raise HTTPException(status_code=404, detail="AA not found")
|
|
return db_aa
|
|
|
|
|
|
@aa_router.put("/assign/{group_id}/{aa_id}", response_model=GroupResponse)
|
|
def assign_accessauth(
|
|
*,
|
|
db: Session = Depends(get_session),
|
|
group_id: int,
|
|
aa_id: int,
|
|
admin: bool = Depends(auth_is_admin),
|
|
):
|
|
db_group = db.get(GroupDB, group_id)
|
|
if db_group is None:
|
|
raise HTTPException(status_code=404, detail="Group not found")
|
|
db_aa = db.get(AccessAuthorizationDB, aa_id)
|
|
if db_aa is None:
|
|
raise HTTPException(status_code=404, detail="AA not found")
|
|
if db_aa in db_group.accessauths:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT, detail="AA already assigned to group"
|
|
)
|
|
db_group.accessauths.append(db_aa)
|
|
return add_and_refresh(db, db_group)
|
|
|
|
|
|
@aa_router.put("/unassign/{group_id}/{aa_id}", response_model=GroupResponse)
|
|
def unassign_accessauth(
|
|
*,
|
|
db: Session = Depends(get_session),
|
|
group_id: int,
|
|
aa_id: int,
|
|
admin: bool = Depends(auth_is_admin),
|
|
):
|
|
db_group = db.get(GroupDB, group_id)
|
|
if db_group is None:
|
|
raise HTTPException(status_code=404, detail="Group not found")
|
|
db_aa = db.get(AccessAuthorizationDB, aa_id)
|
|
if db_aa is None:
|
|
raise HTTPException(status_code=404, detail="AA not found")
|
|
if db_aa not in db_group.accessauths:
|
|
raise HTTPException(status_code=404, detail="AA not assigned to group")
|
|
db_group.accessauths.remove(db_aa)
|
|
return add_and_refresh(db, db_group)
|
|
|
|
|
|
@aa_router.patch("/{aa_id}", response_model=AccessAuthorizationResponse)
|
|
def change_accessauth(
|
|
*,
|
|
db: Session = Depends(get_session),
|
|
aa_id: int,
|
|
aa: AccessAuthorizationUpdate,
|
|
admin: bool = Depends(auth_is_admin),
|
|
):
|
|
db_aa = db.get(AccessAuthorizationDB, aa_id)
|
|
if db_aa is None:
|
|
raise HTTPException(status_code=404, detail="AccessAuthorization not found")
|
|
aa_data = aa.model_dump(exclude_unset=True, exclude_none=True)
|
|
if "oneshot" in aa_data and aa_data["oneshot"] is not None:
|
|
oneshot = OneShotAccess.model_validate(aa_data["oneshot"])
|
|
db_aa.oneshot = oneshot
|
|
aa_data.pop("oneshot")
|
|
db_aa.sqlmodel_update(aa_data)
|
|
return add_and_refresh(db, db_aa)
|
|
|
|
|
|
@aa_router.delete("/{aa_id}")
|
|
def delete_accessauth(
|
|
*,
|
|
db: Session = Depends(get_session),
|
|
aa_id: int,
|
|
admin: bool = Depends(auth_is_admin),
|
|
):
|
|
db_aa = db.get(AccessAuthorizationDB, aa_id)
|
|
if db_aa is None:
|
|
raise HTTPException(status_code=404, detail="AccessAuthorization not found")
|
|
db.delete(db_aa)
|
|
db.commit()
|
|
return {"message": "AccessAuthorization deleted successfully"}
|