Spaces:
Sleeping
Sleeping
File size: 2,972 Bytes
3a19693 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from models import User, UserCreate, UserUpdate, UserResponse, PasswordReset
from auth import hash_password, get_current_admin
router = APIRouter(prefix="/admin", tags=["admin"])
@router.get("/users", response_model=list[UserResponse])
def list_users(
db: Session = Depends(get_db),
_: User = Depends(get_current_admin),
):
return db.query(User).order_by(User.created_at).all()
@router.post("/users", response_model=UserResponse, status_code=201)
def create_user(
user_in: UserCreate,
db: Session = Depends(get_db),
_: User = Depends(get_current_admin),
):
if db.query(User).filter(User.username == user_in.username).first():
raise HTTPException(status_code=400, detail="Username already exists.")
if user_in.email and db.query(User).filter(User.email == user_in.email).first():
raise HTTPException(status_code=400, detail="Email already in use.")
user = User(
username=user_in.username,
full_name=user_in.full_name,
email=user_in.email,
hashed_password=hash_password(user_in.password),
is_admin=user_in.is_admin,
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
return user
@router.put("/users/{user_id}", response_model=UserResponse)
def update_user(
user_id: int,
user_in: UserUpdate,
db: Session = Depends(get_db),
current: User = Depends(get_current_admin),
):
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found.")
for field, value in user_in.model_dump(exclude_unset=True).items():
setattr(user, field, value)
db.commit()
db.refresh(user)
return user
@router.post("/users/{user_id}/reset-password", response_model=UserResponse)
def reset_password(
user_id: int,
body: PasswordReset,
db: Session = Depends(get_db),
_: User = Depends(get_current_admin),
):
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found.")
if not body.new_password or len(body.new_password) < 6:
raise HTTPException(status_code=400, detail="Password must be at least 6 characters.")
user.hashed_password = hash_password(body.new_password)
db.commit()
db.refresh(user)
return user
@router.delete("/users/{user_id}", status_code=204)
def delete_user(
user_id: int,
db: Session = Depends(get_db),
current: User = Depends(get_current_admin),
):
if user_id == current.id:
raise HTTPException(status_code=400, detail="You cannot delete your own account.")
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found.")
db.delete(user)
db.commit()
|