Add admin role with user management (create/delete users)

First registered user becomes admin automatically. Admins see a
"Manage Users" button in the dashboard header that opens a new
/admin page for listing, creating, and deleting users. Backend
enforces admin-only access on /admin/* routes. Startup migration
adds the is_admin column to existing SQLite databases.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-24 19:26:10 +01:00
parent c1371e9c72
commit 0803d86e38
9 changed files with 232 additions and 5 deletions
@@ -27,3 +27,9 @@ def get_current_user(
if user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
return user
def get_current_admin(current_user: models.User = Depends(get_current_user)) -> models.User:
if not current_user.is_admin:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
return current_user
+12 -1
View File
@@ -1,8 +1,9 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import text
from .database import engine, Base
from .routes import users, purchases, stats, history
from .routes import users, purchases, stats, history, admin
Base.metadata.create_all(bind=engine)
@@ -20,6 +21,16 @@ app.include_router(users.router)
app.include_router(purchases.router)
app.include_router(stats.router)
app.include_router(history.router)
app.include_router(admin.router, prefix="/admin")
@app.on_event("startup")
def migrate():
with engine.connect() as conn:
cols = [r[1] for r in conn.execute(text("PRAGMA table_info(users)"))]
if "is_admin" not in cols:
conn.execute(text("ALTER TABLE users ADD COLUMN is_admin BOOLEAN DEFAULT 0"))
conn.commit()
@app.get("/")
+2 -1
View File
@@ -1,4 +1,4 @@
from sqlalchemy import Column, Integer, String, Float, ForeignKey, DateTime
from sqlalchemy import Column, Integer, String, Float, ForeignKey, DateTime, Boolean
from sqlalchemy.orm import relationship
from datetime import datetime
from .database import Base
@@ -10,6 +10,7 @@ class User(Base):
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True, nullable=False)
password = Column(String, nullable=False)
is_admin = Column(Boolean, default=False, nullable=False, server_default='0')
purchases = relationship("Purchase", back_populates="owner", cascade="all, delete")
+57
View File
@@ -0,0 +1,57 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from pydantic import BaseModel
from typing import List
from ..database import get_db
from .. import models
from ..auth import hash_password
from ..dependencies import get_current_admin
router = APIRouter()
class UserOut(BaseModel):
id: int
username: str
is_admin: bool
class Config:
from_attributes = True
class UserCreate(BaseModel):
username: str
password: str
is_admin: bool = False
@router.get("/users", response_model=List[UserOut])
def list_users(db: Session = Depends(get_db), _: models.User = Depends(get_current_admin)):
return db.query(models.User).all()
@router.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def create_user(user_in: UserCreate, db: Session = Depends(get_db), _: models.User = Depends(get_current_admin)):
if db.query(models.User).filter(models.User.username == user_in.username).first():
raise HTTPException(status_code=400, detail="Username already taken")
user = models.User(
username=user_in.username,
password=hash_password(user_in.password),
is_admin=user_in.is_admin,
)
db.add(user)
db.commit()
db.refresh(user)
return user
@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, db: Session = Depends(get_db), current_admin: models.User = Depends(get_current_admin)):
if user_id == current_admin.id:
raise HTTPException(status_code=400, detail="Cannot delete your own account")
user = db.query(models.User).filter(models.User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
db.delete(user)
db.commit()
+4 -1
View File
@@ -17,6 +17,7 @@ class UserCreate(BaseModel):
class Token(BaseModel):
access_token: str
token_type: str
is_admin: bool
@router.post("/register", status_code=status.HTTP_201_CREATED)
@@ -24,9 +25,11 @@ def register(user_in: UserCreate, db: Session = Depends(get_db)):
existing = db.query(models.User).filter(models.User.username == user_in.username).first()
if existing:
raise HTTPException(status_code=400, detail="Username already taken")
no_users_yet = db.query(models.User).first() is None
user = models.User(
username=user_in.username,
password=hash_password(user_in.password),
is_admin=no_users_yet,
)
db.add(user)
db.commit()
@@ -39,4 +42,4 @@ def login(user_in: UserCreate, db: Session = Depends(get_db)):
if not user or not verify_password(user_in.password, user.password):
raise HTTPException(status_code=401, detail="Invalid credentials")
token = create_access_token({"sub": user.username})
return {"access_token": token, "token_type": "bearer"}
return {"access_token": token, "token_type": "bearer", "is_admin": user.is_admin}