derekc
bd2bd43395
- Admin account bootstrapped from ADMIN_EMAIL/ADMIN_PASSWORD env vars on startup - Admin panel: list users, view content, reset passwords, disable/delete accounts - is_admin and is_disabled columns on users table - Disabled accounts blocked at login - README updated with admin setup instructions and panel docs Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
|
|
from auth import get_admin_user, hash_password
|
|
from database import get_db
|
|
from models import Batch, User, Variety
|
|
from schemas import AdminResetPassword, AdminUserOut, BatchOut, VarietyOut
|
|
|
|
router = APIRouter(prefix="/admin", tags=["admin"])
|
|
|
|
|
|
def _user_stats(db: Session, user_id: int) -> dict:
|
|
return {
|
|
"variety_count": db.query(Variety).filter(Variety.user_id == user_id).count(),
|
|
"batch_count": db.query(Batch).filter(Batch.user_id == user_id).count(),
|
|
}
|
|
|
|
|
|
@router.get("/users", response_model=List[AdminUserOut])
|
|
def list_users(db: Session = Depends(get_db), _: User = Depends(get_admin_user)):
|
|
users = db.query(User).order_by(User.created_at).all()
|
|
return [
|
|
AdminUserOut(
|
|
id=u.id, email=u.email, is_admin=u.is_admin, is_disabled=u.is_disabled,
|
|
created_at=u.created_at, **_user_stats(db, u.id)
|
|
)
|
|
for u in users
|
|
]
|
|
|
|
|
|
@router.get("/users/{user_id}/varieties", response_model=List[VarietyOut])
|
|
def user_varieties(user_id: int, db: Session = Depends(get_db), _: User = Depends(get_admin_user)):
|
|
_require_user(db, user_id)
|
|
return db.query(Variety).filter(Variety.user_id == user_id).order_by(Variety.category, Variety.name).all()
|
|
|
|
|
|
@router.get("/users/{user_id}/batches", response_model=List[BatchOut])
|
|
def user_batches(user_id: int, db: Session = Depends(get_db), _: User = Depends(get_admin_user)):
|
|
from sqlalchemy.orm import joinedload
|
|
_require_user(db, user_id)
|
|
return (
|
|
db.query(Batch)
|
|
.options(joinedload(Batch.variety))
|
|
.filter(Batch.user_id == user_id)
|
|
.order_by(Batch.created_at.desc())
|
|
.all()
|
|
)
|
|
|
|
|
|
@router.post("/users/{user_id}/reset-password")
|
|
def reset_password(user_id: int, data: AdminResetPassword, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
|
|
user = _require_user(db, user_id)
|
|
if not data.new_password or len(data.new_password) < 8:
|
|
raise HTTPException(status_code=400, detail="Password must be at least 8 characters")
|
|
user.hashed_password = hash_password(data.new_password)
|
|
db.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.post("/users/{user_id}/disable")
|
|
def toggle_disable(user_id: int, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
|
|
user = _require_user(db, user_id)
|
|
if user.id == admin.id:
|
|
raise HTTPException(status_code=400, detail="Cannot disable your own account")
|
|
if user.is_admin:
|
|
raise HTTPException(status_code=400, detail="Cannot disable another admin account")
|
|
user.is_disabled = not user.is_disabled
|
|
db.commit()
|
|
return {"is_disabled": user.is_disabled}
|
|
|
|
|
|
@router.delete("/users/{user_id}", status_code=204)
|
|
def delete_user(user_id: int, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
|
|
user = _require_user(db, user_id)
|
|
if user.id == admin.id:
|
|
raise HTTPException(status_code=400, detail="Cannot delete your own account")
|
|
if user.is_admin:
|
|
raise HTTPException(status_code=400, detail="Cannot delete another admin account")
|
|
db.delete(user)
|
|
db.commit()
|
|
|
|
|
|
def _require_user(db: Session, user_id: int) -> User:
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return user
|