44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import hash_password, require_super_admin
|
|
from app.database import get_db
|
|
from app.models import AdminUser
|
|
from app.schemas import AdminUserCreate, AdminUserPublic, SubscriberPublic
|
|
from app.models import NewsletterSubscriber
|
|
|
|
router = APIRouter(prefix="/api/admin", tags=["admin-users"])
|
|
|
|
|
|
@router.get("/users", response_model=list[AdminUserPublic])
|
|
def list_users(db: Session = Depends(get_db), _: AdminUser = Depends(require_super_admin)):
|
|
return db.query(AdminUser).order_by(AdminUser.id).all()
|
|
|
|
|
|
@router.post("/users", response_model=AdminUserPublic)
|
|
def create_user(
|
|
body: AdminUserCreate,
|
|
db: Session = Depends(get_db),
|
|
_: AdminUser = Depends(require_super_admin),
|
|
):
|
|
if db.query(AdminUser).filter(AdminUser.email == body.email).first():
|
|
raise HTTPException(status_code=400, detail="Email already registered")
|
|
user = AdminUser(
|
|
email=body.email,
|
|
hashed_password=hash_password(body.password),
|
|
role=body.role,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return AdminUserPublic.model_validate(user)
|
|
|
|
|
|
@router.get("/subscribers", response_model=list[SubscriberPublic])
|
|
def list_subscribers(db: Session = Depends(get_db), _: AdminUser = Depends(require_super_admin)):
|
|
return (
|
|
db.query(NewsletterSubscriber)
|
|
.order_by(NewsletterSubscriber.id.desc())
|
|
.limit(500)
|
|
.all()
|
|
)
|