101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
|
"""Admin-only onboarding: review access requests (Invites), approve/deny, manual add."""
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
|
||
|
|
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
|
||
|
|
from sqlalchemy import select
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from app import email as email_mod
|
||
|
|
from app.auth import current_user
|
||
|
|
from app.db import get_db
|
||
|
|
from app.models import Invite, User
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/api/admin", tags=["admin"])
|
||
|
|
|
||
|
|
|
||
|
|
def admin_user(user: User = Depends(current_user)) -> User:
|
||
|
|
if user.role != "admin":
|
||
|
|
raise HTTPException(status_code=403, detail="Admin only")
|
||
|
|
return user
|
||
|
|
|
||
|
|
|
||
|
|
def _serialize(inv: Invite) -> dict:
|
||
|
|
return {
|
||
|
|
"id": inv.id,
|
||
|
|
"email": inv.email,
|
||
|
|
"status": inv.status,
|
||
|
|
"note": inv.note,
|
||
|
|
"requested_at": inv.requested_at.isoformat() if inv.requested_at else None,
|
||
|
|
"decided_at": inv.decided_at.isoformat() if inv.decided_at else None,
|
||
|
|
"decided_by": inv.decided_by,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/invites")
|
||
|
|
def list_invites(
|
||
|
|
status: str | None = None,
|
||
|
|
_: User = Depends(admin_user),
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
) -> list[dict]:
|
||
|
|
stmt = select(Invite)
|
||
|
|
if status:
|
||
|
|
stmt = stmt.where(Invite.status == status)
|
||
|
|
# Pending first, then most-recently requested.
|
||
|
|
rows = db.execute(stmt.order_by(Invite.requested_at.desc())).scalars().all()
|
||
|
|
order = {"pending": 0, "approved": 1, "denied": 2}
|
||
|
|
rows.sort(key=lambda i: order.get(i.status, 9))
|
||
|
|
return [_serialize(i) for i in rows]
|
||
|
|
|
||
|
|
|
||
|
|
def _decide(db: Session, invite_id: int, admin: User, approved: bool) -> Invite:
|
||
|
|
inv = db.get(Invite, invite_id)
|
||
|
|
if inv is None:
|
||
|
|
raise HTTPException(status_code=404, detail="No such invite")
|
||
|
|
inv.status = "approved" if approved else "denied"
|
||
|
|
inv.decided_at = datetime.now(timezone.utc)
|
||
|
|
inv.decided_by = admin.email
|
||
|
|
db.commit()
|
||
|
|
return inv
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/invites/{invite_id}/approve")
|
||
|
|
def approve_invite(
|
||
|
|
invite_id: int,
|
||
|
|
background: BackgroundTasks,
|
||
|
|
admin: User = Depends(admin_user),
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
) -> dict:
|
||
|
|
inv = _decide(db, invite_id, admin, approved=True)
|
||
|
|
background.add_task(email_mod.send_access_approved, inv.email)
|
||
|
|
return _serialize(inv)
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/invites/{invite_id}/deny")
|
||
|
|
def deny_invite(
|
||
|
|
invite_id: int,
|
||
|
|
admin: User = Depends(admin_user),
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
) -> dict:
|
||
|
|
return _serialize(_decide(db, invite_id, admin, approved=False))
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/invites")
|
||
|
|
def add_invite(
|
||
|
|
payload: dict,
|
||
|
|
admin: User = Depends(admin_user),
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
) -> dict:
|
||
|
|
"""Manually whitelist an email (approved straight away). Convenience for the admin."""
|
||
|
|
email = (payload.get("email") or "").strip().lower()
|
||
|
|
if "@" not in email:
|
||
|
|
raise HTTPException(status_code=400, detail="Enter a valid email address.")
|
||
|
|
inv = db.execute(select(Invite).where(Invite.email == email)).scalar_one_or_none()
|
||
|
|
if inv is None:
|
||
|
|
inv = Invite(email=email)
|
||
|
|
db.add(inv)
|
||
|
|
inv.status = "approved"
|
||
|
|
inv.decided_at = datetime.now(timezone.utc)
|
||
|
|
inv.decided_by = admin.email
|
||
|
|
db.commit()
|
||
|
|
return _serialize(inv)
|