ScarriffleandClaude Opus 5
9be772557e
Das Sitzungs-Token benannte den Benutzer ueber seinen Namen (sub = username). Damit war der Name die Kennung der Sitzung - ein Umbenennen haette jede laufende Anmeldung dieses Benutzers ungueltig gemacht, und deshalb war es gar nicht erst vorgesehen. Beim Anmeldedienst geaenderte Namen liessen sich in Vorrania folglich nirgends nachziehen. Neue Token tragen die ID (sub = "<id>", zusaetzlich uid), und deps.py loest die Sitzung darueber auf. Alles andere haengt ohnehin schon an der ID: Bewegungen, Dashboards, API-Tokens. Der Name ist damit reine Anzeige. Benutzer lassen sich jetzt unter "Benutzer" umbenennen - auch sich selbst, ohne aus der Sitzung zu fliegen. Der Anmeldeweg selbst sucht weiterhin ueber den Namen; das ist das Feld im Formular und soll so bleiben. Bestandsschutz: Token ohne uid werden weiter ueber den Namen aufgeloest, damit beim Update niemand abgemeldet wird. Der Zweig darf nach Ablauf der Token-Lebensdauer weg. Das zusaetzliche uid unterscheidet alt von neu eindeutig - ein Benutzer koennte schlicht "42" heissen. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
109 lines
3.9 KiB
Python
109 lines
3.9 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
||
from sqlalchemy import func
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..database import get_db
|
||
from ..deps import require_admin
|
||
from ..models import Role, User
|
||
from ..schemas import UserCreate, UserOut, UserUpdate
|
||
from ..security import hash_password
|
||
|
||
router = APIRouter(prefix="/users", tags=["users"], dependencies=[Depends(require_admin)])
|
||
|
||
|
||
def _email_normalisiert(roh: str | None) -> str | None:
|
||
"""Leereingabe heisst „keine Adresse"; gespeichert wird immer kleingeschrieben.
|
||
|
||
Die Adresse entscheidet, welches Konto ein SSO-Login beim ersten Mal
|
||
uebernimmt – deshalb darf sie nicht zweimal vergeben sein.
|
||
"""
|
||
if roh is None:
|
||
return None
|
||
sauber = roh.strip().lower()
|
||
return sauber or None
|
||
|
||
|
||
def _email_pruefen(db: Session, email: str | None, ausser_id: int | None = None) -> None:
|
||
if email is None:
|
||
return
|
||
frage = db.query(User).filter(func.lower(User.email) == email)
|
||
if ausser_id is not None:
|
||
frage = frage.filter(User.id != ausser_id)
|
||
if frage.first() is not None:
|
||
raise HTTPException(
|
||
status.HTTP_409_CONFLICT, "Diese E-Mail-Adresse ist bereits vergeben"
|
||
)
|
||
|
||
|
||
@router.get("", response_model=list[UserOut])
|
||
def list_users(db: Session = Depends(get_db)) -> list[User]:
|
||
return db.query(User).order_by(User.username).all()
|
||
|
||
|
||
@router.post("", response_model=UserOut, status_code=status.HTTP_201_CREATED)
|
||
def create_user(payload: UserCreate, db: Session = Depends(get_db)) -> User:
|
||
if db.query(User).filter(User.username == payload.username).first():
|
||
raise HTTPException(status.HTTP_409_CONFLICT, "Benutzername existiert bereits")
|
||
email = _email_normalisiert(payload.email)
|
||
_email_pruefen(db, email)
|
||
user = User(
|
||
username=payload.username,
|
||
password_hash=hash_password(payload.password),
|
||
email=email,
|
||
role=payload.role,
|
||
)
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
return user
|
||
|
||
|
||
@router.patch("/{user_id}", response_model=UserOut)
|
||
def update_user(
|
||
user_id: int, payload: UserUpdate, db: Session = Depends(get_db)
|
||
) -> User:
|
||
user = db.get(User, user_id)
|
||
if user is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Benutzer nicht gefunden")
|
||
if payload.username is not None:
|
||
neuer_name = payload.username.strip()
|
||
if not neuer_name:
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Benutzername darf nicht leer sein")
|
||
if neuer_name != user.username:
|
||
if db.query(User).filter(User.username == neuer_name).first():
|
||
raise HTTPException(status.HTTP_409_CONFLICT, "Benutzername existiert bereits")
|
||
user.username = neuer_name
|
||
if payload.password is not None:
|
||
user.password_hash = hash_password(payload.password)
|
||
if payload.email is not None:
|
||
email = _email_normalisiert(payload.email)
|
||
_email_pruefen(db, email, ausser_id=user.id)
|
||
user.email = email
|
||
if payload.role is not None:
|
||
user.role = payload.role
|
||
db.commit()
|
||
db.refresh(user)
|
||
return user
|
||
|
||
|
||
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
def delete_user(
|
||
user_id: int,
|
||
db: Session = Depends(get_db),
|
||
current_admin: User = Depends(require_admin),
|
||
) -> None:
|
||
user = db.get(User, user_id)
|
||
if user is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Benutzer nicht gefunden")
|
||
if user.id == current_admin.id:
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Du kannst dich nicht selbst löschen")
|
||
# Verhindere das Löschen des letzten Admins.
|
||
if user.role == Role.admin:
|
||
admin_count = db.query(User).filter(User.role == Role.admin).count()
|
||
if admin_count <= 1:
|
||
raise HTTPException(
|
||
status.HTTP_400_BAD_REQUEST, "Der letzte Administrator kann nicht gelöscht werden"
|
||
)
|
||
db.delete(user)
|
||
db.commit()
|